摘要：發(fā)現(xiàn)問題在使用中發(fā)現(xiàn)在設(shè)備頻繁上下線和兩個設(shè)備一樣相互頂替連接的情況下，的和的方法調(diào)用沒有先后順序，如果在這兩個方法里面來記錄設(shè)備上下線狀態(tài)，會造成狀態(tài)不對。因?yàn)橄嗷ロ斕娴那闆r并不多見，因此兩個也可以接受，在性能上并不會造成多大影響。

在moquette使用中發(fā)現(xiàn)在設(shè)備頻繁上下線和兩個設(shè)備ClientId一樣相互頂替連接的情況下，InterceptHandler的onConnect和onConnectionLost的方法調(diào)用沒有先后順序，如果在這兩個方法里面來記錄設(shè)備上下線狀態(tài)，會造成狀態(tài)不對。

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相關(guān)方法的調(diào)用都是在線程池中進(jìn)行的，因此nterceptHandler的onConnect和onConnectionLost的方法調(diào)用先后順序是無法保證的。

在ChannelHandler鏈中添加一個handler，專門處理設(shè)備上線事件，對于相同ClientId的連接已經(jīng)存在時，連接斷開和連接事件強(qiáng)制加上時序。

@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來解決設(shè)備互頂出現(xiàn)的設(shè)備上線和下線回調(diào)時序錯亂的問題
         * 目前解決的方式通過sleep不是太好
         * 解決了多個連接互相頂替出現(xiàn)的問題（有一個連接先連接的情況）
         * */
        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);將斷開連接和處理設(shè)備上線變成先后順序關(guān)系。
3.因?yàn)橄嗷ロ斕娴那闆r并不多見，因此兩個Thread.sleep()也可以接受，在性能上并不會造成多大影響。