发送许多发布消息:正在进行的发布太多错误 [英] Send many publish message: Too many publishes in progress Error

问题描述

这里是 paho 异步客户端:

Here is paho Async client:

    client = new MqttAsyncClient(appProps.getProperty("mqtt.broker"),
            appProps.getProperty("mqtt.clientId"), new MemoryPersistence());
    client.setCallback(this);
    client.connect(null, new IMqttActionListener() {
        @Override
        public void onSuccess(IMqttToken imt) {
            try {
                client.subscribe(Constants.internalTopics, Constants.internalTopicQOS);
            } catch (MqttException ex) {
                ex.printStackTrace();
            }
        }

        @Override
        public void onFailure(IMqttToken imt, Throwable thrwbl) {
            thrwbl.printStackTrace();
        }
    });

我在这里循环发送消息:

Here I am sending messages in loop:

        while (iterator.hasNext()) {
            try {
               client.publish("user/" + userId + "/downstream", mqttMessage);
            } catch(Exception ex) {
                ex.printStackTrace();
            }
        }

错误:

Too many publishes in progress (32202)
    at org.eclipse.paho.client.mqttv3.internal.ClientState.send(ClientState.java:436)
    at org.eclipse.paho.client.mqttv3.internal.ClientComms.internalSend(ClientComms.java:121)
    at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:139)
    at org.eclipse.paho.client.mqttv3.MqttAsyncClient.publish(MqttAsyncClient.java:858)
    at org.eclipse.paho.client.mqttv3.MqttAsyncClient.publish(MqttAsyncClient.java:836)

我正在使用 Rabbitmq

推荐答案

看source 对于 Paho 客户端，它看起来像默认的最大数量任何给定时间的机上消息都是 10.

Looking at the source for the Paho client it looks like the default maximum number of inflight messages at any given time is 10.

因此，考虑到您的发布循环有多紧凑，它只需要在网络层稍微放慢一点速度，并且您最终会在任何给定时间发送超过 10 条消息.如果您尝试以大于 0 的 QOS 发送，这只会变得更糟.

So given how tight your publish loop is it will only take a small slow down in the network layer and your going to end up with more than 10 messages in the process of being sent at any given time. This will only get worse if you try to send at a QOS greater than 0.

您可以在传递给 client.connect()MQTTConnectionsOptions 对象上使用 setMaxInflight(int n) 方法更改默认值> 方法.

You can change the default with the setMaxInflight(int n) method on the MQTTConnectionsOptions object that is passed to the client.connect() method.

我建议你尝试找到一个合适的值.

I suggest you experiment to find a suitable value.

这篇关于发送许多发布消息:正在进行的发布太多错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！