moquette改造笔记(五)：设备连接频繁上下线或者相互顶替出现的设备上下线状态错乱问题...

发现问题

在moquette使用中发现在设备频繁上下线和两个设备ClientId一样相互顶替连接的情况下&＃xff0c;InterceptHandler的onConnect和onConnectionLost的方法调用没有先后顺序&＃xff0c;如果在这两个方法里面来记录设备上下线状态&＃xff0c;会造成状态不对。

io.moquette.spi.impl.ProtocolProcessor中的processConnect(Channel channel, MqttConnectMessage msg)部分代码如下

ConnectionDescriptor descriptor &＃61; new ConnectionDescriptor(clientId, channel, cleanSession);final ConnectionDescriptor existing &＃61; this.connectionDescriptors.addConnection(descriptor);if (existing !&＃61; null) {LOG.info("Client ID is being used in an existing connection, force to be closed. CId&＃61;{}", clientId);existing.abort();this.connectionDescriptors.removeConnection(existing);this.connectionDescriptors.addConnection(descriptor);}initializeKeepAliveTimeout(channel, msg, clientId);storeWillMessage(msg, clientId);if (!sendAck(descriptor, msg, clientId)) {channel.close().addListener(CLOSE_ON_FAILURE);return;}m_interceptor.notifyClientConnected(msg);

可以看到existing.abort();后会m_interceptor.notifyClientConnected(msg); 先断开原来的连接&＃xff0c;然后接着通知上线。由于Netty本身就是异步的&＃xff0c;再加上InterceptHandler相关方法的调用都是在线程池中进行的&＃xff0c;因此nterceptHandler的onConnect和onConnectionLost的方法调用先后顺序是无法保证的。

解决方法

在ChannelHandler链中添加一个handler&＃xff0c;专门处理设备上线事件&＃xff0c;对于相同ClientId的连接已经存在时&＃xff0c;连接断开和连接事件强制加上时序。

&＃64;Sharable
public class AbrotExistConnectionMqttHandler extends ChannelInboundHandlerAdapter {private static final Logger LOG &＃61; LoggerFactory.getLogger(AbrotExistConnectionMqttHandler.class);private final ProtocolProcessor m_processor;private static final ReentrantLock[] locks &＃61; new ReentrantLock[8];static {for (int i &＃61; 0; i }

解释&＃xff1a;
1.通过ReentrantLock lock &＃61; locks[Math.abs(clientId.hashCode()) % locks.length];来保证相同的ClientId的连接都会获得同一个锁
2.通过两次Thread.sleep(50);将断开连接和处理设备上线变成先后顺序关系。
3.因为相互顶替的情况并不多见&＃xff0c;因此两个Thread.sleep()也可以接受&＃xff0c;在性能上并不会造成多大影响。