发布者使用 qos 2 发布得到经纪人或订阅者的确认

Question

我对qos有点困惑，我读到有关 qos 的信息如果 qos 设置为 2，则 broker/client 将使用四步握手传递消息一次。

所以 qos 2 确认消息发布在代理上，而不是订阅者（客户端）收到。要么订阅者收到消息或者

对于确认，我们应该需要建立应用程序，例如发布者将发布带有主题的消息，例如 "DATA"，并将订阅一个主题，例如 "ACK"，订阅者需要发布关于主题的确认"ACK" 收到关于主题 "DATA"

的消息

我创建了一个 java class 用于发布数据，另一个 class 用于订阅出版商

在下面的代码中，我尝试在 qos 2 和 deliveryComplete 函数中尝试 getMessage() 时出现异常，而当我尝试 qos 0 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 从上面 link

QoS2, Exactly once：消息总是恰好传递一次。消息必须存储在发送方本地，直到发送方收到消息已被接收方发布的确认。存储消息以防必须再次发送消息。 QoS2 是最安全但最慢的传输模式。

Answer 1

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

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

客户端上满足 QOS 合同的所有存储都应由您正在使用的 MQTT 库处理（在本例中为 Paho）

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

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

发布者使用 qos 2 发布得到经纪人或订阅者的确认

publisher publishing with qos 2 get the acknowledge from broker or subcriber

java

eclipse

mqtt

mosquitto

paho