如何停止 java 方法，直到它收到来自 MQTT 代理的消息答案

【问题标题】：How to stop a java method until it receives a message from a MQTT broker如何停止 java 方法，直到它收到来自 MQTT 代理的消息
【发布时间】：2018-08-11 16:41:41
【问题描述】：

我有一个包含 2 个方法的类，每个方法分别发布和订阅一个通用 MQTT 代理。

public class operations{
    public void fetch(){
    // do something

     mqttConn.mqttSubscriberFetch(MQTTtopic);

    // do something
   }

    public void store(){
   // do something

    mqttConn.mqttSubscriberStore(MQTTtopic);

   // do something
   }

}

方法获取的MQTT方法如下：

public class mqttConnectionFetch implements  MqttCallback{

    MqttClient client;

    String topic;

    public mqttConnectionFetch() {
    }


    public void mqttSubscriberFetch(String topic) {

        final String MQTT_BROKER = "tcp://localhost:1883" ;
        final String CLIENT_ID = "fetch";
        try {
            client = new MqttClient(MQTT_BROKER, CLIENT_ID);
            client.connect();
            client.setCallback(this);
            client.subscribe(topic);
            MqttMessage message = new MqttMessage();

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }



    @Override
    public void connectionLost(Throwable cause) {
        // TODO Auto-generated method stub

    }

    @Override
    public void messageArrived(String topic, MqttMessage message)
            throws Exception {

        System.out.println("the message received is "+message);

        if(message.toString().equals("Processed")) {
            MqttPublisherFetch("Processed", "operations/fetch");

        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // TODO Auto-generated method stub

    }


    public void MqttPublisherFetch(String message, String topic) {
        final String MQTT_BROKER = "tcp://localhost:1883" ;
        final String CLIENT_ID = "store";
        try {
            client = new MqttClient(MQTT_BROKER, CLIENT_ID);
            client.connect();
            createMqttMessage(message,topic);
            client.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private void createMqttMessage(String message, String topic) throws MqttException {
        MqttMessage publishMessage = new MqttMessage();
        publishMessage.setPayload(message.getBytes());
        client.publish(topic, publishMessage);

    }


}

现在我正在尝试拥有这样一个功能，即每当我的 fetch 方法订阅一个主题时，如果来自代理的消息是“处理中”，它应该进入等待状态。虽然 store 方法应该正常工作。当收到的消息是“已处理”时，提取应该再次开始工作。

我用普通的 java wait() 和 start() 试过这个，但我没有得到想要的输出。有人可以帮我揭开这个神秘面纱吗？

【问题讨论】：

为什么不简单地使用观察者/监听者模式，并在状态变化通知时调用适当的方法？
首先，几乎没有充分的理由在同一个进程中运行多个 MQTT 客户端，一个对发布和订阅都很好
我不确定我是否理解您想要的内容... fetch 客户端按书面形式订阅 MQTT 服务器并等待，仅在收到来自代理的消息时才唤醒。如果它从代理接收到“正在处理”消息，它可以忽略/记录它并返回等待（这是它已经要做的）；如果它收到“已处理”消息，它可以去处理它 - 然后回去等待下一个。您想如何改变这种行为？
如果远程客户端关闭并且您从未收到响应，会发生什么？
您似乎正在解决一些问题，但正如已经以非常低效的方式指出的那样。你能描述一下为什么你需要这样做吗？基本上你能描述你试图解决的更高层次的问题吗？

标签： java mqtt paho

【解决方案1】：

据我了解，存储方法包括几个步骤。其中一个步骤是通过 MQTT 发送消息并等待响应。从 store 方法的客户端的角度来看，它是同步的，即客户端只有在整个处理（包括通过 MQTT 的异步发送/接收）完成后才会收到响应。

您要在这里解决的是一个经典问题，即一个线程需要等待另一个线程发生某种情况。有很多方法可以实现这一点，只需检查How to make a Java thread wait for another thread's output? 中提出的选项数量即可。

最简单的方法是使用CountDownLatch。我将使用fetch 方法，因为这是您为其提供代码的方法。

首先您将修改mqttSubscriberFetch 以创建一个内部CountDownLatch 对象：

private CountDownLatch processingFinishedLatch;

public void mqttSubscriberFetch(String topic) {

    final String MQTT_BROKER = "tcp://localhost:1883" ;
    final String CLIENT_ID = "fetch";
    try {
        client = new MqttClient(MQTT_BROKER, CLIENT_ID);
        client.connect();
        client.setCallback(this);
        client.subscribe(topic);
        MqttMessage message = new MqttMessage();
        processingFinishedLatch = new CountDownLatch(1);

    } catch (MqttException e) {
        e.printStackTrace();
    }
}

那么你需要在消息接收回调中触发这个信号：

@Override
public void messageArrived(String topic, MqttMessage message)
        throws Exception {

    System.out.println("the message received is "+message);

    if(message.toString().equals("Processed")) {
        MqttPublisherFetch("Processed", "operations/fetch");
        this.processingFinishedLatch.countDown();
    }

}

还需要为mqttConnectionFetch的客户端提供一种等待latch变为零的方法：

void waitForProcessingToFinish() throws InterruptedException {
     this.processingFinishedLatch.await();
}

fetch 方法应该像这样使用它：

public void fetch(){
// do something

mqttConn.mqttSubscriberFetch(MQTTtopic);

// do everything that is needed to initiate processing

mqttConn.waitForProcessingToFinish()

// do something

}

正在执行fetch 的线程将在waitForProcessingToFinish 中等待，直到latch 达到零，这将在适当的消息到来时发生。

当消息永远不会出现时，可以轻松修改此方法以解决问题。只需使用await with some timeout：

boolean waitForProcessingToFinish(long timeout,
        TimeUnit unit) throws InterruptedException {
     return this.processingFinishedLatch.await();
}

fetch 应该检查返回值，如果发生超时，可能会向调用者返回错误。

请注意，此解决方案的缺点是处理fetch 的线程将一直很忙。如果这是用于处理传入 HTTP 请求的线程池中的线程，这可能会限制您的服务器可以同时处理的请求数。有一些方法可以缓解这种情况，例如 async servlets、RxJava 或 Spring WebFlux。这个问题的实际解决方案很大程度上取决于您用于 REST API 实现的技术。

【讨论】：