摘要：發現問題在使用中發現在設備頻繁上下線和兩個設備一樣相互頂替連接的情況下，的和的方法調用沒有先后順序，如果在這兩個方法里面來記錄設備上下線狀態，會造成狀態不對。因為相互頂替的情況并不多見，因此兩個也可以接受，在性能上并不會造成多大影響。

在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()也可以接受，在性能上并不會造成多大影響。