数据库优化

smqtt-common/src/main/java/io/github/quickmsg/common/channel/MqttChannel.java

@@ -41,8 +41,6 @@ public class MqttChannel {
 
     private ChannelStatus status;
 
-    private long activeTime;
-
     private long authTime;
 
     private long connectTime;
@@ -101,7 +99,6 @@ public class MqttChannel {
         mqttChannel.setReplyMqttMessageMap(new ConcurrentHashMap<>());
         mqttChannel.setMqttMessageSink(new MqttMessageSink());
         mqttChannel.setQos2MsgCache(new ConcurrentHashMap<>());
-        mqttChannel.setActiveTime(System.currentTimeMillis());
         mqttChannel.setConnection(connection);
         mqttChannel.setStatus(ChannelStatus.INIT);
         mqttChannel.setAddress(connection.address().toString());

smqtt-core/src/main/java/io/github/quickmsg/core/protocol/ConnectProtocol.java

@@ -66,15 +66,15 @@ public class ConnectProtocol implements Protocol<MqttConnectMessage> {
             byte mqttVersion = (byte) mqttConnectVariableHeader.version();
             PasswordAuthentication passwordAuthentication = mqttReceiveContext.getPasswordAuthentication();
             /*check clientIdentifier exist*/
+            MqttChannel existMqttChannel = channelRegistry.get(clientIdentifier);
             if (mqttReceiveContext.getConfiguration().getConnectModel() == ConnectModel.UNIQUE) {
-                if (channelRegistry.exists(clientIdentifier)) {
+                if (existMqttChannel != null && existMqttChannel.getStatus() == ChannelStatus.ONLINE) {
                     return mqttChannel.write(
                             MqttMessageBuilder.buildConnectAck(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, mqttVersion),
                             false).then(mqttChannel.close());
                 }
             } else {
-                MqttChannel existMqttChannel = channelRegistry.get(clientIdentifier);
-                if (existMqttChannel != null) {
+                if (existMqttChannel != null && existMqttChannel.getStatus() == ChannelStatus.ONLINE) {
                     if (System.currentTimeMillis() - existMqttChannel.getConnectTime() > (mqttReceiveContext.getConfiguration().getNotKickSecond() * 1000)) {
                         existMqttChannel.close().subscribe();
                     } else {