✨ mica-mqtt client onSubscribed 改为异步

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientProcessor.java

@@ -189,13 +189,17 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
 		paddingSubscribe.onSubAckReceived();
 		clientSession.removePaddingSubscribe(messageId);
 		clientSession.addSubscriptionList(subscribedList);
-		try {
-			subscribedList.forEach(clientSubscription -> {
-				clientSubscription.getListener().onSubscribed(clientSubscription.getTopicFilter(), clientSubscription.getMqttQoS());
-			});
-		} catch (Throwable e) {
-			logger.error("MQTT SubscriptionList:{} subscribed onSubscribed event error.", subscribedList);
-		}
+		// 触发已经监听的事件
+		subscribedList.forEach(clientSubscription -> {
+			IMqttClientMessageListener subscriptionListener = clientSubscription.getListener();
+			try {
+				executor.execute(() ->
+					subscriptionListener.onSubscribed(clientSubscription.getTopicFilter(), clientSubscription.getMqttQoS())
+				);
+			} catch (Throwable e) {
+				logger.error("MQTT SubscriptionList:{} subscribed onSubscribed event error.", subscribedList);
+			}
+		});
 	}
 
 	@Override

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientCreator.java

@@ -28,6 +28,7 @@ import org.tio.client.intf.ClientAioHandler;
 import org.tio.client.intf.ClientAioListener;
 import org.tio.core.Node;
 import org.tio.core.ssl.SslConfig;
+import org.tio.utils.Threads;
 import org.tio.utils.hutool.StrUtil;
 import org.tio.utils.thread.pool.DefaultThreadFactory;
 
@@ -393,7 +394,7 @@ public final class MqttClientCreator {
 		if (this.messageIdGenerator == null) {
 			this.messageIdGenerator = new DefaultMqttClientMessageIdGenerator();
 		}
-		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2, DefaultThreadFactory.getInstance("MqttClient"));
+		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(Threads.CORE_POOL_SIZE, DefaultThreadFactory.getInstance("MqttClient"));
 		IMqttClientProcessor processor = new DefaultMqttClientProcessor(this, executor);
 		// 4. 初始化 mqtt 处理器
 		ClientAioHandler clientAioHandler = new MqttClientAioHandler(this, processor);

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/server/MqttServerCreator.java

@@ -40,6 +40,7 @@ import org.tio.server.ServerTioConfig;
 import org.tio.server.TioServer;
 import org.tio.server.intf.ServerAioHandler;
 import org.tio.server.intf.ServerAioListener;
+import org.tio.utils.Threads;
 import org.tio.utils.hutool.StrUtil;
 import org.tio.utils.thread.pool.DefaultThreadFactory;
 
@@ -460,7 +461,7 @@ public class MqttServerCreator {
 		if (this.connectStatusListener == null) {
 			this.connectStatusListener = new DefaultMqttConnectStatusListener();
 		}
-		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2, DefaultThreadFactory.getInstance("MqttServer"));
+		ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(Threads.CORE_POOL_SIZE, DefaultThreadFactory.getInstance("MqttServer"));
 		DefaultMqttServerProcessor serverProcessor = new DefaultMqttServerProcessor(this, executor);
 		// 1. 处理消息
 		ServerAioHandler handler = new MqttServerAioHandler(this, serverProcessor);