logo标签列表

使用Mqttclient异步和并发处理多个消息

javamultithreadingmqttpaho
4
我正在开发一个程序，它会消费来自MQTT主题的消息，我的目标是能异步地消费和处理多个消息。我正在使用Eclipse客户端： https://www.eclipse.org/paho/files/javadoc/org/eclipse/paho/client/mqttv3/MqttClient.html https://www.eclipse.org/paho/files/javadoc/org/eclipse/paho/client/mqttv3/MqttAsyncClient.html 问题在于，多个消息不能同时被处理，它们都在同一个线程中执行。我不太理解使用MqttClient和MqttAsyncClient之间的区别。Javadoc上说：
MqttClient
轻量级客户端，使用阻塞方法与MQTT服务器通信，直到操作完成。
MqttAsyncClient
轻量级客户端，使用非阻塞方法与MQTT服务器通信，允许操作在后台运行。
我并不是很清楚使用“subscribe”方法或“setCallback”方法之间的区别。只有使用“subscribe”方法，您才能声明多个监听器： setCallback
设置回调侦听器以处理异步发生的事件。 订阅 订阅主题...
我尝试同时发送了十条消息。我的测试如下：
public class FooListener implements IMqttMessageListener {
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("Thread [ " + Thread.currentThread().getName() + 
                "], Topic[ "+ topic + "],  Message [" + message +"] ");
    }
}

public class FooCallbackListener implements MqttCallback {

    @Override
    public void connectionLost(Throwable e) {
        e.printStackTrace();
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //TODO:emtpy
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("Thread [ " + Thread.currentThread().getName() + 
                "], Topic[ "+ topic + "],  Message [" + message +"] ");
    }

}

MqttClient和订阅：
public class FooMqttClient {

    public static void main(String[] args) {
        MqttConnectOptions connOpt = new MqttConnectOptions();
        connOpt.setCleanSession(true);
        connOpt.setKeepAliveInterval(30);
        String serverUri = "tcp://iot.eclipse.org:1883";
        String clientId = UUID.randomUUID().toString();

        try {
            MqttClient myClient = new MqttClient(serverUri, clientId);
            myClient.connect(connOpt);
            myClient.subscribe("topic/foo", new FooListener());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

结果：

Thread [ MQTT Call: 7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[topic/foo],  Message [Foo 0]  
Thread [ MQTT Call:7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 1]  
Thread [ MQTT Call: 7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 2]  
Thread [ MQTT Call:7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 3]  
Thread [ MQTT Call: 7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 4]  
Thread [ MQTT Call:7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 5]  
Thread [ MQTT Call: 7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 6]  
Thread [ MQTT Call:7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 7]  
Thread [ MQTT Call: 7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 8]  
Thread [ MQTT Call:7bb52a01-03db-4870-8921-8d6432f2fe27], Topic[ topic/foo],  Message [Foo 9]

MqttClient和setCallback:
public class FooMqttCallbackClient {

    public static void main(String[] args) {
        MqttConnectOptions connOpt = new MqttConnectOptions();
        connOpt.setCleanSession(true);
        connOpt.setKeepAliveInterval(30);
        String serverUri = "tcp://iot.eclipse.org:1883";
        String clientId = UUID.randomUUID().toString();

        try {
            MqttClient myClient = new MqttClient(serverUri, clientId);
            myClient.connect(connOpt);
            myClient.subscribe("topic/foo");
            myClient.setCallback(new FooCallbackListener());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

结果：

Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 0]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 1]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 2]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 3]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 4]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 5]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 6]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 7]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 8]  
Thread [ MQTT Call: 3d571fd4-906d-4a1f-b06f-a2f7d43b95b2], Topic[ topic/foo],  Message [Foo 9]

MqttAsyncClient和subscribe：
public class FooAsyncMqttClient {
    public static void main(String[] args) {
        MqttConnectOptions connOpt = new MqttConnectOptions();
        connOpt.setCleanSession(true);
        connOpt.setKeepAliveInterval(30);
        String serverUri = "tcp://iot.eclipse.org:1883";
        String clientId = UUID.randomUUID().toString();

        try {
            MqttAsyncClient myClient = new MqttAsyncClient(serverUri, clientId);
            myClient.connect(connOpt);
            Thread.sleep(1000);
            myClient.subscribe("topic/foo", 1, new FooListener());
        } catch (MqttException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

结果：

Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 0]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 1]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 2]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 3]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 4]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 5]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 6]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 7]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 8]  
Thread [ MQTT Call: bbc21897-3c38-4f96-8dea-74522250afee], Topic[ topic/foo],  Message [Foo 9]

MqttAsyncClient 和 setCallback
public class FooAsyncMqttCallbackClient {

    public static void main(String[] args) {
        MqttConnectOptions connOpt = new MqttConnectOptions();
        connOpt.setCleanSession(true);
        connOpt.setKeepAliveInterval(30);
        String serverUri = "tcp://iot.eclipse.org:1883";
        String clientId = UUID.randomUUID().toString();

        try {
            MqttAsyncClient myClient = new MqttAsyncClient(serverUri, clientId);
            myClient.connect(connOpt);
            Thread.sleep(1000);
            myClient.subscribe("topic/foo", 1);
            myClient.setCallback(new FooCallbackListener());
        } catch (MqttException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }
}

结果：

Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 0]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 1]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 2]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 3]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 4]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 5]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 6]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 7]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 8]  
Thread [ MQTT Call: 75b827c9-34fe-4e2f-a723-cf0f277b91d6], Topic[ topic/foo],  Message [Foo 9]

在我所有的测试中，监听器都是在同一个线程中执行而不是并发执行。我该怎么做才能同时且并发地处理消息？MqttClient和MqttAsyncClient有什么区别？ 解决方案： 
public class FooExecutorListener implements IMqttMessageListener {

    private ExecutorService pool = Executors.newFixedThreadPool(10);

    class MessageHandler implements Runnable {
        MqttMessage message;
        String topic;

        public MessageHandler(String topic, MqttMessage message) {
            this.message = message;
            this.topic = topic;
        }

        public void run() {
            System.out.println("Thread [ " + Thread.currentThread().getName() + 
                    "], Topic[ "+ topic + "],  Message [" + message +"] ");
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        pool.execute(new MessageHandler(topic, message));
    }

}

结果：

Thread [ pool-2-thread-1], Topic[ topic/foo],  Message [Foo 0] 
Thread [ pool-2-thread-2], Topic[ topic/foo],  Message [Foo 1] 
Thread [ pool-2-thread-5], Topic[ topic/foo],  Message [Foo 4] 
Thread [ pool-2-thread-4], Topic[ topic/foo],  Message [Foo 3] 
Thread [ pool-2-thread-7], Topic[ topic/foo],  Message [Foo 6] 
Thread [ pool-2-thread-6], Topic[ topic/foo],  Message [Foo 5] 
Thread [ pool-2-thread-8], Topic[ topic/foo],  Message [Foo 7] 
Thread [ pool-2-thread-3], Topic[ topic/foo],  Message [Foo 2] 
Thread [ pool-2-thread-1], Topic[ topic/foo],  Message [Foo 10] 
Thread [ pool-2-thread-2], Topic[ topic/foo],  Message [Foo 11] 
Thread [ pool-2-thread-5], Topic[ topic/foo],  Message [Foo 12] 
Thread [ pool-2-thread-5], Topic[ topic/foo],  Message [Foo 13] 
Thread [ pool-2-thread-5], Topic[ topic/foo],  Message [Foo 14] 
Thread [ pool-2-thread-9], Topic[ topic/foo],  Message [Foo 8] 
Thread [ pool-2-thread-10], Topic[ topic/foo],  Message [Foo 9]
1个回答
6

两个版本的客户端之间的区别在于连接/发布而不是订阅。异步版本将连接和发布而不会阻塞。

在两种情况下，订阅处理都在后台网络线程上处理。

如果您想并行处理传入的消息，则需要实现自己的线程池并将传入的消息分配到该池中。

最简单的方法是使用Java的ExecutorService类。例如：

public class FooListener implements IMqttMessageListener {

  ExecutorService pool = new ExecutorService(10);

  class MessageHandler implements Runnable {
    MqttMessage message;
    String topic;

    public MessageHandler(String topic; MqttMessage message) {
      this.message = message;
    }

    public void run() {
      //process message
    }
  }

  @Override
  public void messageArrived(String topic, MqttMessage message) throws Exception {
    pool.execute(new MessageHandler(topic,message));
  }
}
你有没有关于在8核系统中实现最佳吞吐量的理想线程池大小的参考？我们的系统可以轻松通过HTTP服务器每秒提供500个请求，但是使用mosquitto代理和上述示例的客户端，线程池大小为30时，性能不佳。任何帮助将不胜感激。 - Siddharth Trikha
回答这个问题的唯一方法是进行测试/基准测试，因为它完全取决于每条消息的处理方式。 - hardillb
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接