Threads param added

src/main/java/org/redisson/Config.java

@@ -30,6 +30,11 @@ public class Config {
 
     private SingleConnectionConfig singleConnectionConfig;
 
+    /**
+     * Threads amount shared between all redis node clients
+     */
+    private int threads = 0; // 0 = current_processors_amount * 2
+
     /**
      * Redis key/value codec. JsonJacksonCodec used by default
      */
@@ -44,6 +49,7 @@ public class Config {
             oldConf.setCodec(new JsonJacksonCodec());
         }
 
+        setThreads(oldConf.getThreads());
         setCodec(oldConf.getCodec());
         if (oldConf.getSingleConnectionConfig() != null) {
             setSingleConnectionConfig(new SingleConnectionConfig(oldConf.getSingleConnectionConfig()));
@@ -99,4 +105,13 @@ public class Config {
         this.masterSlaveConnectionConfig = masterSlaveConnectionConfig;
     }
 
+    public int getThreads() {
+        return threads;
+    }
+
+    public Config setThreads(int threads) {
+        this.threads = threads;
+        return this;
+    }
+
 }

src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java

@@ -51,7 +51,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private RedisCodec codec;
 
-    private final EventLoopGroup group = new NioEventLoopGroup();
+    private EventLoopGroup group;
 
     private final List<ConnectionEntry> slaveConnections = new ArrayList<ConnectionEntry>();
     private final Queue<RedisConnection> masterConnections = new ConcurrentLinkedQueue<RedisConnection>();
@@ -75,6 +75,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
     }
 
     void init(MasterSlaveConnectionConfig config, Config cfg) {
+        this.group = new NioEventLoopGroup(cfg.getThreads());
         this.config = config;
         for (URI address : this.config.getSlaveAddresses()) {
             RedisClient client = new RedisClient(group, address.getHost(), address.getPort());
@@ -240,6 +241,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
     }
 
     public void shutdown() {
+        masterClient.shutdown();
         for (RedisClient client : slaveClients) {
             client.shutdown();
         }