Add BrokerConstants.NETTY_WORKER_THREADS

moquette-io · Apr 8, 2019 · c520b9f · c520b9f
1 parent 1ee744f
commit c520b9f
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -71,6 +71,7 @@ public final class BrokerConstants {
     public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll";
     public static final String NETTY_MAX_BYTES_PROPERTY_NAME = "netty.mqtt.message_size";
     public static final int DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE = 8092;
+    public static final String NETTY_WORKER_THREADS = "netty.worker_threads";
     public static final String IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME = "immediate_buffer_flush";
     public static final String METRICS_ENABLE_PROPERTY_NAME = "use_metrics";
     public static final String METRICS_LIBRATO_EMAIL_PROPERTY_NAME = "metrics.librato.email";

diff --git a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
@@ -146,17 +146,22 @@ public void initialize(NewNettyMQTTHandler mqttHandler, IConfig props, ISslConte
         nettyChannelTimeoutSeconds = props.intProp(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, 10);
         maxBytesInMessage = props.intProp(BrokerConstants.NETTY_MAX_BYTES_PROPERTY_NAME,
                 BrokerConstants.DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE);
+        int nettyWorkerThreads = props.intProp(BrokerConstants.NETTY_WORKER_THREADS, 0);
+
+        bossGroup = new EpollEventLoopGroup();
+        if (nettyWorkerThreads == 0) {
+            workerGroup = new EpollEventLoopGroup();
+        }
+        else {
+            workerGroup = new EpollEventLoopGroup(nettyWorkerThreads);
+        }
 
         boolean epoll = props.boolProp(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, false);
         if (epoll) {
             LOG.info("Netty is using Epoll");
-            bossGroup = new EpollEventLoopGroup();
-            workerGroup = new EpollEventLoopGroup();
             channelClass = EpollServerSocketChannel.class;
         } else {
             LOG.info("Netty is using NIO");
-            bossGroup = new NioEventLoopGroup();
-            workerGroup = new NioEventLoopGroup();
             channelClass = NioServerSocketChannel.class;
         }