java - 使用 QoS 2 进行发布的发布者获得代理或订阅者的确认

Question

我对服务质量有点困惑，我读到有关 QoS 的内容是如果 qos 设置为 2，则代理/客户端将通过使用四步握手仅传递一次消息。

所以 qos 2 确认消息已在代理上发布，而不是由订阅者(客户端)接收。或者订阅者收到消息或

对于确认，我们应该需要建立应用程序，例如发布者将发布带有主题(例如“DATA”)的消息，并将订阅主题(例如“ACK”)，并且订阅者需要发布主题“ACK”的确认，该消息是收到主题“数据”

我创建了一个用于发布数据的java类和另一个用于订阅的类出版商

在下面的代码中，我尝试在 qos 2 上发布，并且在 DeliveryComplete 函数中，当我尝试使用 qos 0 getMessage() 时尝试 getMessage() 时出现异常，但没有给出任何异常。

public class PublishMe implements MqttCallback{
    MqttClient myClient;
    MqttClient myClientPublish;
    MqttConnectOptions connOpt;
    MqttConnectOptions connOptPublish;
    static final String BROKER_URL = "tcp://Ehydromet-PC:1883";

    static Boolean msgACK=false;    
    public static void main(String[] args) {
        PublishMe smc = new PublishMe();
        smc.runClient();
    }
    @Override
    public void connectionLost(Throwable t) {
        System.out.println("Connection lost!");
    }

        @Override
        public void messageArrived(String string, MqttMessage message) throws Exception {
                System.out.println("-------------------------------------------------");
        System.out.println("| Topic:" + string);
        System.out.println("| Message: " + new String(message.getPayload()));
        System.out.println("-------------------------------------------------");

        }
/**
     * 
     * deliveryComplete
     * This callback is invoked when a message published by this client
     * is successfully received by the broker.
     * 
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
       try{
        System.out.println("Message delivered successfully to topic : \"" + token.getMessage().toString() + "\".");
       }catch(Exception ex){
        System.out.println(ex.getCause()+" -- "+ex.getLocalizedMessage()+" -- "+ex.getMessage()+" -- " );      
               }

       }

    public void runClient() {
        connOpt = new MqttConnectOptions();
        connOpt.setCleanSession(false);
        connOpt.setKeepAliveInterval(0);

                connOptPublish= new MqttConnectOptions();
        connOptPublish.setCleanSession(false);
        connOptPublish.setKeepAliveInterval(0);

// Connect to Broker
        try {
            myClient = new MqttClient(BROKER_URL, "pahomqttpublish11");
            myClient.setCallback(this);
            myClient.connect(connOpt);

                        myClientPublish= new MqttClient(BROKER_URL, "pahomqttpublish42");
            myClientPublish.setCallback(this);
            myClientPublish.connect(connOptPublish);

        } catch (MqttException e) {
            e.printStackTrace();
            System.exit(-1);
        }

        System.out.println("Connected to " + BROKER_URL);

        String myTopic = "sample";
//                String myTopic = "receiveDATA2";
                MqttTopic topic = myClientPublish.getTopic(myTopic);

        // publish messages if publisher
        if (publisher) {

                    int i=1;
            while(true){
                                String pubMsg = "sample msg "+i;

                MqttMessage message = new MqttMessage(pubMsg.getBytes());
                                System.out.println(message);
                                message.setQos(2);
                                message.setRetained(false);

                                // Publish the message
                                MqttDeliveryToken token = null;
                                try {
                    // publish message to broker
                    token = topic.publish(message);
                    // Wait until the message has been delivered to the broker
                    token.waitForCompletion();
                                        msgACK=false;
                    Thread.sleep(100);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }           
        }
    }


}

下面是订阅者

public class Mqttsample implements MqttCallback{
    MqttClient myClient;
    MqttClient myClientPublish;
    MqttConnectOptions connOpt;
MqttConnectOptions connOptPublish;
    static final String BROKER_URL = "tcp://Ehydromet-PC:1883";
    // the following two flags control whether this example is a publisher, a subscriber or both
    static final Boolean subscriber = true;
    static final Boolean publisher = true;
        public static void main(String[] args) {


        Mqttsample smc = new Mqttsample();
        smc.runClient();
    }
       @Override
    public void connectionLost(Throwable t) {
        System.out.println("Connection lost!");
        // code to reconnect to the broker would go here if desired
    }

        @Override
        public void messageArrived(String string, MqttMessage message) throws Exception {
        //throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
                System.out.println("| Topic:" + string+"| Message: " + new String(message.getPayload()));

        }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try{
            System.out.println("Pub complete" + new String(token.getMessage().getPayload()));
        }
        catch(Exception ex ){
            System.out.println("delivery Error "+ex.getMessage());
        }
       }



    public void runClient() {
        connOpt = new MqttConnectOptions();
        connOpt.setCleanSession(false);
        connOpt.setKeepAliveInterval(0);

                connOptPublish= new MqttConnectOptions();
        connOptPublish.setCleanSession(false);
        connOptPublish.setKeepAliveInterval(0);

// Connect to Broker
        try {
            myClient = new MqttClient(BROKER_URL, "pahomqttpublish");
            myClient.setCallback(this);
            myClient.connect(connOpt);

                        myClientPublish= new MqttClient(BROKER_URL, "pahomqttsubscribe");
            myClientPublish.setCallback(this);
            myClientPublish.connect(connOptPublish);

        } catch (MqttException e) {
            e.printStackTrace();
            System.exit(-1);
        }

        System.out.println("Connected to " + BROKER_URL);


        // subscribe to topic if subscriber
        if (subscriber) {
            try {
                            //String myTopicACK = M2MIO_DOMAIN + "/" + "ACK" + "/" + M2MIO_THING;
                            String myTopicACK = "sample";
                           // MqttTopic topicACK = myClient.getTopic(myTopicACK);
                int subQoS = 2;

                myClient.subscribe(myTopicACK, subQoS);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
//                 


    }


}

我如何确保订阅者已收到消息，我应该在发布者代码中实现什么。

http://www.eclipse.org/paho/files/mqttdoc/Cclient/qos.html从上面的链接

QoS2，恰好一次:消息始终只传递一次。该消息必须存储在发送方本地，直到发送方收到消息已由接收方发布的确认。该消息将被存储，以防必须再次发送该消息。 QoS2 是最安全但最慢的传输模式。

Answer 1

正如您所确定的，较高的 QOS 级别仅描述客户端(发布者或订阅者)与代理之间的消息传递，而不是端到端发布者到订阅者之间的消息传递。

这是经过深思熟虑的，因为作为发布/订阅协议(protocol)，无法知道某个主题可能有多少订阅者。可以是 0 到 n 之间的任意数字。此外，发布者和订阅者可以在不同的 QOS 级别与主题进行交互(发布者可以以 QOS 2 发布，订阅者可以以 QOS 0 订阅)。消息还可以作为保留消息发布，这样最后保留的消息将始终传递到新订阅的客户端。

客户端上满足 QOS 契约(Contract)的所有存储都应由您正在使用的 MQTT 库处理(在本例中为 Paho)

deliveryComplete 回调仅指示发布者已完成向代理发送消息。还有doc表示如果消息已传递，token.getMessage() 将返回 null，这将解释您提到的异常(我必须在这里猜测，因为您没有包含该异常)。

如果您的应用程序架构确实需要端到端的消息确认，那么您将需要实现类似于您所描述的内容。但为了确保它正常工作，您应该在消息的有效负载中包含一个消息 ID，并且确认消息应该包含此 ID，并且可能包含某种识别哪个订阅者正在回复的方法，以确保您知道谁收到了该消息。我使用这样的东西的唯一原因是如果确认消息有时间要求。如果时间不是相关因素，那么请考虑使用持久 session 来确保订阅客户端在发布时断开连接并重新连接时将消息传送到订阅客户端。