fix: fix bug for mset cross slot (miss values except cluster mode)

Signed-off-by: xujie <mikawudi@qq.com>

redisson/src/main/java/org/redisson/RedissonBuckets.java

@@ -150,7 +150,7 @@ public class RedissonBuckets implements RBuckets {
                 }
                 return params.toArray();
             }
-        }, buckets.keySet().toArray(new String[]{}));
+        }, buckets.keySet().toArray(new String[]{}), buckets);
     }
 
 }

redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java

@@ -28,6 +28,7 @@ import org.redisson.liveobject.core.RedissonObjectBuilder;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 
 /**
  *
@@ -115,7 +116,9 @@ public interface CommandAsyncExecutor {
     ByteBuf encodeMapValue(Codec codec, Object value);
 
     <T, R> RFuture<R> readBatchedAsync(Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String... keys);
-    
+
     <T, R> RFuture<R> writeBatchedAsync(Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String... keys);
     
+    <T, R> RFuture<R> writeBatchedAsync(Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String[] keys, Map<String, ?> valueMap);
+    
 }

redisson/src/main/java/org/redisson/command/CommandAsyncService.java

@@ -588,20 +588,35 @@ public class CommandAsyncService implements CommandAsyncExecutor {
     
     @Override
     public <T, R> RFuture<R> readBatchedAsync(Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String... keys) {
-        return executeBatchedAsync(true, codec, command, callback, keys);
+        return executeBatchedAsync(true, codec, command, callback, keys, null);
     }
-    
+
     @Override
     public <T, R> RFuture<R> writeBatchedAsync(Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String... keys) {
-        return executeBatchedAsync(false, codec, command, callback, keys);
+        return executeBatchedAsync(false, codec, command, callback, keys, null);
+    }
+    
+    @Override
+    public <T, R> RFuture<R> writeBatchedAsync(Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String[] keys, Map<String, ?> valueMap) {
+        return executeBatchedAsync(false, codec, command, callback, keys, valueMap);
     }
     
-    private <T, R> RFuture<R> executeBatchedAsync(boolean readOnly, Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String... keys) {
+    private <T, R> RFuture<R> executeBatchedAsync(boolean readOnly, Codec codec, RedisCommand<T> command, SlotCallback<T, R> callback, String[] keys, Map<String, ?> valueMap) {
         if (!connectionManager.isClusterMode()) {
+            List<Object> params = null;
+            if (valueMap != null) {
+                params = new ArrayList<>(keys.length * 2);
+                for (String key : keys) {
+                    params.add(key);
+                    params.add(valueMap.get(key));
+                }
+            } else {
+                params = Arrays.asList(keys);
+            }
             if (readOnly) {
-                return readAsync((String) null, codec, command, keys);
+                return readAsync((String) null, codec, command, params.toArray());
             }
-            return writeAsync((String) null, codec, command, keys);
+            return writeAsync((String) null, codec, command, params.toArray());
         }
 
         Map<MasterSlaveEntry, Map<Integer, List<String>>> entry2keys = Arrays.stream(keys).collect(