摘要：性能正在發(fā)布過(guò)多的消息系統(tǒng)性能，注意請(qǐng)使用單線程的萬(wàn)條毫秒萬(wàn)條毫秒萬(wàn)毫秒萬(wàn)條毫秒多線程的正在發(fā)布過(guò)多的消息問(wèn)題異常信息正在進(jìn)行過(guò)多的發(fā)布解決辦法消息發(fā)送發(fā)送限流用多帶帶的一個(gè)線程來(lái)完成消息的推送不用這個(gè)，使用就沒(méi)有事增加的值反思筆者出現(xiàn)這個(gè)錯(cuò)

org.eclipse.paho.client.mqttv3

2.2 GHz Intel Core i7 mac系統(tǒng)

publish性能，注意請(qǐng)使用單線程的 mqttclinet

1萬(wàn)條 341毫秒
4萬(wàn)條 1163毫秒
5萬(wàn) 1450毫秒
10萬(wàn)條 2700毫秒

多線程的 mqttclinet?MQTT(32202): 正在發(fā)布過(guò)多的消息 問(wèn)題

[15:07:21]: publish failed, message: aaaa
正在進(jìn)行過(guò)多的發(fā)布 (32202)
 at org.eclipse.paho.client.mqttv3.internal.ClientState.send(ClientState.java:496)
 at org.eclipse.paho.client.mqttv3.internal.ClientComms.internalSend(ClientComms.java:132)
 at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:156)
 at org.eclipse.paho.client.mqttv3.MqttAsyncClient.publish(MqttAsyncClient.java:1027)
 at org.eclipse.paho.client.mqttv3.MqttClient.publish(MqttClient.java:399)
 at io.communet.ichater.emq.util.MqttUtil.publishMsg(MqttUtil.java:171)
 at io.communet.ichater.emq.util.MqttUtil.publishMsg(MqttUtil.java:161)
 at io.communet.ichater.emq.sub.MqttSendMsgEventSubscribe.onEvent(MqttSendMsgEventSubscribe.java:28)
 at java.lang.reflect.Method.invoke(Native Method)
 at java.lang.reflect.Method.invoke(Method.java:372)
 at org.greenrobot.eventbus.EventBus.invokeSubscriber(EventBus.java:507)
 at org.greenrobot.eventbus.EventBus.invokeSubscriber(EventBus.java:501)
 at org.greenrobot.eventbus.AsyncPoster.run(AsyncPoster.java:46)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
 at java.lang.Thread.run(Thread.java:818)

消息發(fā)送發(fā)送限流

用多帶帶的一個(gè)線程來(lái)完成 MQ 消息的推送 (不用這個(gè)MqttAsyncClient ，使用MqttClient 就沒(méi)有事)

options.setMaxInflight(1000)?增加?actualInFlight?的值;

筆者出現(xiàn)這個(gè)錯(cuò)誤是因?yàn)槭褂?EventBus, 之前使用多帶帶線程的 Handler 是沒(méi)有問(wèn)題的, 調(diào)查發(fā)現(xiàn), 使用 EventBus 是新建線程運(yùn)行的, 而 Handler 是多帶帶一個(gè)線程.
所以當(dāng)發(fā)送大量消息的時(shí)候, EventBus 幾乎是同一個(gè)點(diǎn)發(fā)出去, 就會(huì)造成這個(gè)錯(cuò)誤

根據(jù)堆棧信息找到報(bào)錯(cuò)地方

if (actualInFlight >= this.maxInflight) {
    //@TRACE 613= sending {0} msgs at max inflight window
    log.fine(CLASS_NAME, methodName, "613", new Object[]{new Integer(actualInFlight)});

    throw new MqttException(MqttException.REASON_CODE_MAX_INFLIGHT);
}

其中?actualInFlight?如下

// processed until the inflight window has space. 
if (actualInFlight < this.maxInflight) {
    // The in flight window is not full so process the 
    // first message in the queue
    result = (MqttWireMessage)pendingMessages.elementAt(0);
    pendingMessages.removeElementAt(0);
    actualInFlight++;
    
    //@TRACE 623=+1 actualInFlight={0}
    log.fine(CLASS_NAME,methodName,"623",new Object[]{new Integer(actualInFlight)});
}

從?pendingMessages?中取出消息時(shí),?actualInFlight?加 1,?maxInflight?可以自己設(shè)定, 默認(rèn)值為 10.

public class ClientState {
  ...
  volatile private Vector pendingMessages;
  ...
}

在?ClientState?中：

public void send(MqttWireMessage message, MqttToken token) throws MqttException {
        ... 
        if (message instanceof MqttPublish) {
            synchronized (queueLock) {
                if (actualInFlight >= this.maxInflight) {
                    //@TRACE 613= sending {0} msgs at max inflight window
                    log.fine(CLASS_NAME, methodName, "613", new Object[]{new Integer(actualInFlight)});

                    throw new MqttException(MqttException.REASON_CODE_MAX_INFLIGHT);
                }
                
                MqttMessage innerMessage = ((MqttPublish) message).getMessage();
                //@TRACE 628=pending publish key={0} qos={1} message={2}
                log.fine(CLASS_NAME,methodName,"628", new Object[]{new Integer(message.getMessageId()), new Integer(innerMessage.getQos()), message});

                switch(innerMessage.getQos()) {
                    case 2:
                        outboundQoS2.put(new Integer(message.getMessageId()), message);
                        persistence.put(getSendPersistenceKey(message), (MqttPublish) message);
                        break;
                    case 1:
                        outboundQoS1.put(new Integer(message.getMessageId()), message);
                        persistence.put(getSendPersistenceKey(message), (MqttPublish) message);
                        break;
                }
                tokenStore.saveToken(token, message);
                pendingMessages.addElement(message);
                queueLock.notifyAll();
            }
        } else {
        ...
        }
    }

可以看到?pendingMessages?中添加元素的時(shí)候并沒(méi)有做?qos?類型的判斷

private void decrementInFlight() {
    final String methodName = "decrementInFlight";
    synchronized (queueLock) {
        actualInFlight--;
        //@TRACE 646=-1 actualInFlight={0}
        log.fine(CLASS_NAME,methodName,"646",new Object[]{new Integer(actualInFlight)});
            
        if (!checkQuiesceLock()) {
            queueLock.notifyAll();
        }
    }
}

當(dāng)收到消息反饋時(shí)?actualInFlight?減 1.