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

摘要：发现问题在使用中发现在设备频繁上下线和两个设备一样相互顶替连接的情况下，的和的方法调用没有先后顺序，如果在这两个方法里面来记录设备上下线状态，会造成状态不对。因为相互顶替的情况并不多见，因此两个也可以接受，在性能上并不会造成多大影响。

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

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

ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession);
        final ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
        if (existing != null) {
            LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", 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); 先断开原来的连接，然后接着通知上线。由于Netty本身就是异步的，再加上InterceptHandler相关方法的调用都是在线程池中进行的，因此nterceptHandler的onConnect和onConnectionLost的方法调用先后顺序是无法保证的。

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

@Sharable
public class AbrotExistConnectionMqttHandler extends ChannelInboundHandlerAdapter {
    private static final Logger LOG = LoggerFactory.getLogger(AbrotExistConnectionMqttHandler.class);
    private final ProtocolProcessor m_processor;

    private static final ReentrantLock[] locks = new ReentrantLock[8];

    static {
        for (int i = 0; i < locks.length; i++) {
            locks[i] = new ReentrantLock();
        }
    }

    public AbrotExistConnectionMqttHandler(ProtocolProcessor m_processor) {
        this.m_processor = m_processor;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
        MqttMessage msg = (MqttMessage) message;
        MqttMessageType messageType = msg.fixedHeader().messageType();
        LOG.debug("Processing MQTT message, type: {}", messageType);

        if (messageType != MqttMessageType.CONNECT) {
            super.channelRead(ctx, message);
            return;
        }

        MqttConnectMessage connectMessage = (MqttConnectMessage) msg;
        String clientId = connectMessage.payload().clientIdentifier();

        /**
         * 通过锁和sleep来解决设备互顶出现的设备上线和下线回调时序错乱的问题
         * 目前解决的方式通过sleep不是太好
         * 解决了多个连接互相顶替出现的问题（有一个连接先连接的情况）
         * */
        ReentrantLock lock = locks[Math.abs(clientId.hashCode()) % locks.length];
        lock.lock();
        try {
            if (!m_processor.isConnected(clientId)) {
                super.channelRead(ctx, message);
                return;
            }
            m_processor.abortIfExist(clientId);
            Thread.sleep(50);
            super.channelRead(ctx, message);
            Thread.sleep(30);
        } catch (Exception ex) {
            ex.printStackTrace();
            super.channelRead(ctx, message);
        } finally {
            lock.unlock();
        }

    }
}

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