如何理解PUBLISH报文处理

这期内容当中小编将会给大家带来有关如何理解PUBLISH报文处理，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。

先简单说明一下，对于mqtt是个双向通信的过程，也就是说，他既允许client向broker发布消息，同时也允许broker向client发布消息

public void processPublish(Channel channel, MqttPublishMessage msg) {    final MqttQoS qos = msg.fixedHeader().qosLevel();    final String clientId = NettyUtils.clientID(channel);    LOG.info("Processing PUBLISH message. CId={}, topic={}, messageId={}, qos={}", clientId,            msg.variableHeader().topicName(), msg.variableHeader().packetId(), qos);    switch (qos) {        case AT_MOST_ONCE:            this.qos0PublishHandler.receivedPublishQos0(channel, msg);            break;        case AT_LEAST_ONCE:            this.qos1PublishHandler.receivedPublishQos1(channel, msg);            break;        case EXACTLY_ONCE:            this.qos2PublishHandler.receivedPublishQos2(channel, msg);            break;        default:            LOG.error("Unknown QoS-Type:{}", qos);            break;    }}

根据发布的消息的qos，用不同的QosPublishHandler来处理，QosPublishHandler有三个具体实现，分别是Qos0PublishHandler，Qos1PublishHandler，Qos2PublishHandler.
这里面先讲解Qos1PublishHandler的处理

void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {    // verify if topic can be write    final Topic topic = new Topic(msg.variableHeader().topicName());    String clientID = NettyUtils.clientID(channel);    String username = NettyUtils.userName(channel);    if (!m_authorizator.canWrite(topic, username, clientID)) {        LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);        return;    }    final int messageID = msg.variableHeader().packetId();    // route message to subscribers    IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);    toStoreMsg.setClientID(clientID);    this.publisher.publish3Subscribers(toStoreMsg, topic, messageID);    sendPubAck(clientID, messageID);    if (msg.fixedHeader().isRetain()) {        if (!msg.payload().isReadable()) {            m_messagesStore.cleanRetained(topic);        } else {            // before wasn't stored            m_messagesStore.storeRetained(topic, toStoreMsg);        }    }    // 修改publish消息，slice出的ByteBuf对象，原文中存在内存泄漏    MoquetteMessage moquetteMessage = new MoquetteMessage(msg.fixedHeader(), msg.variableHeader(), msg.content());    m_interceptor.notifyTopicPublished(moquetteMessage, clientID, username);    msg.content().release();}    1.鉴权，该client下的username是否有对该topic发布消息（对topic写）的权限。    2.创建一个IMessagesStore.StoredMessage，同时把消息推送给所有该对该消息的订阅者。        if (LOG.isTraceEnabled()) {        LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +                "subscriptionTree={}", pubMsg.getClientID(), topic, messageID, DebugUtils.payload2Str(pubMsg.getPayload()),            subscriptions.dumpTree());    } else if(LOG.isInfoEnabled()){        LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", pubMsg.getClientID(), topic,            messageID);    }    publish3Subscribers(pubMsg, topic);判断是否是跟踪模式，如果是的话，会把当前所有的订阅关系打印到日志，由于这个需要遍历topic树，消耗比较大，所以是可配置的，在moquette.cof里面配置。核心的处理逻辑在下面，接着往下看    void publish3Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic) {    List topicMatchingSubscriptions = subscriptions.matches(topic);    final String topic1 = pubMsg.getTopic();    final MqttQoS publishingQos = pubMsg.getQos();    final ByteBuf origPayload = pubMsg.getPayload();    for (final Subscription sub : topicMatchingSubscriptions) {        MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);        ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());        boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());//TODO move all this logic into messageSender, which puts into the flightZone only the messages that pull out of the queue.        if (targetIsActive) {            if(LOG.isDebugEnabled()){                LOG.debug("Sending PUBLISH message to active subscriber. CId={}, topicFilter={}, qos={}",                    sub.getClientId(), sub.getTopicFilter(), qos);            }            // we need to retain because duplicate only copy r/w indexes and don't retain() causing            // refCnt = 0            ByteBuf payload = origPayload.retainedDuplicate();            MqttPublishMessage publishMsg;            if (qos != MqttQoS.AT_MOST_ONCE) {                // QoS 1 or 2                int messageId = targetSession.inFlightAckWaiting(pubMsg);                // set the PacketIdentifier only for QoS > 0                publishMsg = notRetainedPublishWithMessageId(topic1, qos, payload, messageId);            } else {                publishMsg = notRetainedPublish(topic1, qos, payload);            }            this.messageSender.sendPublish(targetSession, publishMsg);        } else {            if (!targetSession.isCleanSession()) {                if(LOG.isDebugEnabled()){                    LOG.debug("Storing pending PUBLISH inactive message. CId={}, topicFilter={}, qos={}",                        sub.getClientId(), sub.getTopicFilter(), qos);                }                // store the message in targetSession queue to deliver                targetSession.enqueue(pubMsg);            }        }    }}

大概分为以下几步
2.1.根据topic找出匹配的订阅集合list，这里面由于涉及到比较大的计算，所以单独讲解

         public List matches(Topic topic) {    Queue tokenQueue = new LinkedBlockingDeque<>(topic.getTokens());    List matchingSubs = new ArrayList<>();    subscriptions.get().matches(tokenQueue, matchingSubs);  // 客户端使用带通配符的主题过滤器请求订阅时，客户端的订阅可能会重复，因此发布的消息可能会匹配多个过滤器。对于这种情        //况，服务端必须将消息分发给所有订阅匹配的QoS等级最高的客户端。服务端之后可以按照订阅的QoS等级，分发消息的副本给每        //一个匹配的订阅者。    Map subsForClient = new HashMap<>();    for (ClientTopicCouple matchingCouple : matchingSubs) {        Subscription existingSub = subsForClient.get(matchingCouple.clientID);        Subscription sub = this.subscriptionsStore.getSubscription(matchingCouple);        if (sub == null) {            // if the m_sessionStore hasn't the sub because the client disconnected            continue;        }        // update the selected subscriptions if not present or if has a greater qos        if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) {            subsForClient.put(matchingCouple.clientID, sub);//注意这里最终存入的从session里面获取的订阅，而不是从topic目录里面获取的，是因为，有可能client当时并不在线或者该订阅的qos等级变化了        }    }    return new ArrayList<>(subsForClient.values());}可以看的出来，会先创建一个队列，存储topic的层级比如/a/b/c，队列里面就会有三个运输[c,b,a]   这里面之所以要用到队列而不是，list就是因为后面进行匹配的时候需要确保先从第一个层级开始匹配，而不是最后一个void matches(Queue tokens, List matchingSubs) {    Token t = tokens.poll();    // check if t is null <=> tokens finished    if (t == null) {        matchingSubs.addAll(m_subscriptions);        // check if it has got a MULTI child and add its subscriptions        for (TreeNode n : m_children) {            if (n.getToken() == Token.MULTI || n.getToken() == Token.SINGLE) {                matchingSubs.addAll(n.subscriptions());            }        }        return;    }    // we are on MULTI, than add subscriptions and return    if (m_token == Token.MULTI) {        matchingSubs.addAll(m_subscriptions);        return;    }    for (TreeNode n : m_children) {        if (n.getToken().match(t)) {            // Create a copy of token, else if navigate 2 sibling it            // consumes 2 elements on the queue instead of one            n.matches(new LinkedBlockingQueue<>(tokens), matchingSubs);            // TODO don't create a copy n.matches(tokens, matchingSubs);        }    }}

这段代码不好理解，涉及到迭代topic树io.moquette.spi.impl.subscriptions.TreeNode，下面以一个图说明
另外对topic匹配规则不熟悉的同学可以看一下这里https://github.com/mcxiaoke/mqtt/blob/master/mqtt/04-OperationalBehavior.md
假如有A,B,C,D, E五个client，其中A订阅了/test/#，B订阅了/test/hello/#，C 订阅了/test/hello/beijing,
D订阅了/test/+/hello，现在E向topic-name为/test/hello/shanghai发布了一条消息请问哪几个client应该收到这条消息。
向画出topic树如下（请原谅我手画）

先分析 E发布消息的整个过程：
2.1.1./test/hello/shanghai被放入queue，取出来的顺序依次为test，hello，shanghai
2.1.2.第一轮先匹配test，test不为空，RootNode(其实是null)不为#，执行到遍历RootNode下的子节点，RootNode先的子节点只有一个，test，test.equals(test)，然后当前treenode变为test，
2.1.3.从queue里面取出hello，hello不为空，test不为#，遍历test这个treenode的子节点，test有三个子节点，分别是（#，+，hello）
2.1. 3.1 子节点是#，# .mathcs(hello)，当前节点是#，然后从队列里面取出shanghai，shanghai不为空，#为#，当前迭代终止，节点A匹配放入匹配的list，
2.1.3.2 子节点是+，+ .mathcs(hello)当前节点是+，然后从队列里面取出shanghai（这里可能有的同学有疑问，为什么还能取出shanghai呢，因为进行下一级迭代的时候是new的新的queue），上海不为空，+不为#所以不匹配，接着匹配+这个treenode的子节点，+只有一个子节点
2.1. 3.2.1 当前节点是hello，hello.mathcs(shanghai)不成立，迭代终止，所以D不会放入匹配的list
1.3.3 子节点是hello，hello.equals(hello)，当前节点变成hello，然后从队列里面取出shanghai，shanghai不为空，hello不为#，遍历hello这个treenode下的子节点，hello先有两个子节点，分别是（#，beijing），
2.1.3.3.1 子节点是#，# .mathcs(hello)，当前节点是#，从队列里面取出的是空，所以直接会走第一个if分支，并且把B放入匹配的list，并且退出方法。
2.1. 3.3.2 子节点是beijing，beijing.equals（shanghai）不成立，退出迭代
最终能够匹配成功的只有A,B这两个client
也就说说能够被成功匹配，要么/test/hello/shanghai的每一层级都能成功匹配，"+"能够够任意的单个层级，或者某一个层级是"#"（分别对应上面的两个if分支）
到此位置，有topic-name匹配由topic-filters组成的topic树的整个过程分析完成了。下面接着回到上面的publish3Subscribers的这个方法讲解匹配出client之后的动作

2.2 遍历找出的匹配的client,确定qos，qos取订阅请求要求的qos和发布的消息自身的qos的最小值，这个是mqtt协议自身规定的，之所以这样规定是因为，订阅的时候其基本单位是某个topic，订阅者只能订阅一个topic，而不能订阅一个消息，而发布消息的基本单位是一个消息，一个topic下可以有很多个消息，不同的消息可以有不同的qos，所以这里面，真正的qos是由订阅方和发布方共同决定的，出于性能的考虑，去最小的qos
2.3 根据连接描述符，判断是否client依然在线，如果不在线，且客户端要求保留会话，则把消息保存到该client的session的BlockingQueue，以待client再次上线之后，再次发送给该client，这个对应着在建立连接的时候有一个republish的动作，具体看https://blog.51cto.com/13579730/2073630的 第10步
2.3.如果在线，根据qos做不同的处理，如果qos是0，比较简单，之间发送，qos是1或者2，则会先把消息放入outboundFlightZone，产生一个messageId，再通过PersistentQueueMessageSender进行发送
io.moquette.spi.impl.PersistentQueueMessageSender#sendPublish，具体的分发逻辑比较简单，这里不详细讲解。类之间的关系是
ProtocolProcessor--》qos1PublishHandler--》MessagesPublisher--》PersistentQueueMessageSender，基本的逻辑就是通过ConnectionDescriptorStore进行发送，对于发送失败的要求保存会话的qos1或者2消息，将会继续保留，知道重复成功

到此publish3Subscribers方法即发送的核心逻辑讲解完了，让我们回到io.moquette.spi.impl.Qos1PublishHandler#receivedPublishQos1这个方法，

3.发送PUBACK消息，
4.如果是retain消息，但是有没有paylod，将该topic下的retain消息清除掉，可以理解成是客户端主动要求清除的，因为它发送了一个空的消息，如果有payload，则存储retain消息，对于保留消息，详细看这里https://github.com/mcxiaoke/mqtt/blob/master/mqtt/0303-PUBLISH.md 我简单总结一下，1.每个topic下永远只会存储一条retain消息，如果发送了两条，那么后面的将会将前面的覆盖；2.如果客户端发送了一个零字节的retain消息，broker将会清理调该topic下的retain消息，因为broker不会存储零字节的retain消息；3.服务端的保留消息不是会话状态的组成部分。服务端应该保留那种消息直到客户端删除它。4.当client收到broker发送的一个retain消息是，可以理解成这是client新建立的一个订阅的第一条消息。
5.唤醒拦截器

到此moquette对PUBLISH报文的处理讲解完了，这里面只讲解了qos1的处理，是因为qos0处理比较简单，而qos2我们没什么应用场景。另外这里说明一下比较容易混淆的概念cleanSession与retain消息
1.retain消息并不是session的一部分，它不与client挂钩，而是与topic-filter挂钩。也就是说当发布这个retain消息的clientsession不存在了，但是retain消息依然存在，除非有client主动删除它，
2.对于要求保留会话的client，会存在一个broker主动重新发送消息的过程，这个动作实在client重新建立连接的时候，具体看这里https://blog.51cto.com/13579730/2073630 的第10步，这是因为broker有责任对要求保留会话的client重新发送qos1与qos2消息
3.对于client发布订阅的时候，broker也会有一个主动发送消息的过程，具体看这里https://blog.51cto.com/13579730/2073914 的第8步
4.qos1消息和qos2是在消息发送失败或者，client不在线的时候，存储到clientsession里面的一个BlockingQueue里面的，而retain消息是在，broker收到的时候直接存储到IMessagesStore进行存储的，其底层是个Map

上述就是小编为大家分享的如何理解PUBLISH报文处理了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注行业资讯频道。