From 74b283743545f643b710f7afe4709c82d8f541ad Mon Sep 17 00:00:00 2001
From: zhouhao <zh.sqy@qq.com>
Date: Thu, 15 Sep 2022 11:06:00 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BD=BF=E7=94=A8parallel=E7=BA=BF=E7=A8=8B?=
 =?UTF-8?q?=E6=B1=A0=E5=A4=84=E7=90=86=E6=95=B0=E6=8D=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../network/mqtt/gateway/device/MqttClientDeviceGateway.java    | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java
index b47b8234..8c0fe83c 100644
--- a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java
+++ b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java
@@ -21,6 +21,7 @@ import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.supports.server.DecodedClientMessageHandler;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
@@ -117,6 +118,7 @@ public class MqttClientDeviceGateway extends AbstractDeviceGateway {
                             );
                     })
                     .then()
+                    .subscribeOn(Schedulers.parallel())
                     .onErrorResume((err) -> {
                         log.error("处理MQTT消息失败:{}", mqttMessage, err);
                         return Mono.empty();