mqtt协议-broker之moqutte源码研究二之SUBSCRIBE报文处理

Posted 2020-10-25

这一篇开始讲解moqutte对SUBSCRIBE报文的处理

代码不复杂
public void processSubscribe(Channel channel, MqttSubscribeMessage msg) {
String clientID = NettyUtils.clientID(channel);//从channel里面获取clientId，具体原理看下文
int messageID = messageId(msg);
LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID);

    RunningSubscription executionKey = new RunningSubscription(clientID, messageID);
    SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED);
    if (currentStatus != null) {
        LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}",
            clientID, messageID);
        return;
    }
    String username = NettyUtils.userName(channel);
    List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg);
    MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID);
    if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) {
        LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " +
            "messageId={}", clientID, messageID);
        return;
    }

    LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics);

    List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID);

    // save session, persist subscriptions from session
    for (Subscription subscription : newSubscriptions) {
        subscriptions.add(subscription.asClientTopicCouple());
    }

    LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID);
    channel.writeAndFlush(ackMessage);

    // fire the persisted messages in session
    for (Subscription subscription : newSubscriptions) {
        publishRetainedMessagesInSession(subscription, username);
    }

    boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED);
    if (!success) {
        LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID);
    }
}

1.channel里面为什么会存在clientid呢？这个问题也可以这样描述，当连接建立之后，client发布消息的时候，netty接收到socket里面的数据之后，他怎么知道是哪个client的数据呢？这里面就需要确定client与channel的映射关系。moquette是这么做的，
在处理CONNECT的第5步，详见http://blog.51cto.com/13579730/2073630的时候会做如下处理
private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, final String clientId) {
    int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
    LOG.info("Configuring connection. CId={}", clientId);
    NettyUtils.keepAlive(channel, keepAlive);
    // session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession());
    NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
    // used to track the client in the subscription and publishing phases.
    // session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID());
    NettyUtils.clientID(channel, clientId);
    int idleTime = Math.round(keepAlive * 1.5f);
    setIdleTime(channel.pipeline(), idleTime);

    if(LOG.isDebugEnabled()){
        LOG.debug("The connection has been configured CId={}, keepAlive={}, cleanSession={}, idleTime={}",
                clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
    }
}

    这里面有一步NettyUtils.clientID(channel, clientId);这个不起眼的方法做了将channel与clientId映射的动作，接着跟踪
     public static void clientID(Channel channel, String clientID) {
    channel.attr(NettyUtils.ATTR_KEY_CLIENTID).set(clientID);
}
    原来是把clientId作为一个属性存到了channel里面，因为channel是集成AttributeMap的，所以可以这么做。

只要有channel与clientId的映射关系，就好说了，这也就是为什么moquette的NettyMQTTHandler是这样处理的

    @Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
    MqttMessage msg = (MqttMessage) message;
    MqttMessageType messageType = msg.fixedHeader().messageType();
    if(LOG.isDebugEnabled())
        LOG.debug("Processing MQTT message, type={}", messageType);
    try {
        switch (messageType) {
            case CONNECT:
                m_processor.processConnect(ctx.channel(), (MqttConnectMessage) msg);
                break;
            case SUBSCRIBE:
                m_processor.processSubscribe(ctx.channel(), (MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:
                m_processor.processUnsubscribe(ctx.channel(), (MqttUnsubscribeMessage) msg);
                break;
            case PUBLISH:
                m_processor.processPublish(ctx.channel(), (MqttPublishMessage) msg);
                break;
            case PUBREC:
                m_processor.processPubRec(ctx.channel(), msg);
                break;
            case PUBCOMP:
                m_processor.processPubComp(ctx.channel(), msg);
                break;
            case PUBREL:
                m_processor.processPubRel(ctx.channel(), msg);
                break;
            case DISCONNECT:
                m_processor.processDisconnect(ctx.channel());
                break;
            case PUBACK:
                m_processor.processPubAck(ctx.channel(), (MqttPubAckMessage) msg);
                break;
            case PINGREQ:
                MqttFixedHeader pingHeader = new MqttFixedHeader(
                        MqttMessageType.PINGRESP,
                        false,
                        AT_MOST_ONCE,
                        false,
                        0);
                MqttMessage pingResp = new MqttMessage(pingHeader);
                ctx.writeAndFlush(pingResp);
                break;
            default:
                LOG.error("Unkonwn MessageType:{}", messageType);
                break;

哪个tcp-socket对应哪个channel由netty负责处理，当client发送数据的时候，netty负责从ChannelHandlerContext取出channel传给相应的业务自定义的handler进行处理。

2.创建一个正在运行中的RunningSubscription对象，之所以要创建这个对象，是为了防止重复订阅，同时到存储了所有的RunningSubscription的ConcurrentMap里面查询所有已经存在这个对象，如果存在，说明是重复的订阅包，则不处理，这里面调用了putIfAbsent方法，同时重写了RunningSubscription的equals方法。packetId和clientID相同时代表是相同的RunningSubscription
3.从channel里面取出用户名，验证该client下的该username是否有权利读取该topic（订阅该topic）的权限，这里贴一下相关的代码进行讲解

    rivate List<MqttTopicSubscription> doVerify(String clientID, String username, MqttSubscribeMessage msg) {
    ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
    List<MqttTopicSubscription> ackTopics = new ArrayList<>();

    final int messageId = messageId(msg);
    for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
        Topic topic = new Topic(req.topicName());
        if (!m_authorizator.canRead(topic, username, clientSession.clientID)) {
            // send SUBACK with 0x80, the user hasn‘t credentials to read the topic
            LOG.error("Client does not have read permissions on the topic CId={}, username={}, messageId={}, " +
                "topic={}", clientID, username, messageId, topic);
            ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE));
        } else {
            MqttQoS qos;
            if (topic.isValid()) {
                LOG.info("Client will be subscribed to the topic CId={}, username={}, messageId={}, topic={}",
                    clientID, username, messageId, topic);
                qos = req.qualityOfService();
            } else {
                LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", clientID,
                    username, messageId, topic);
                qos = FAILURE;
            }
            ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));
        }
    }
    return ackTopics;
}

从报文的payload里面取出所有的订阅请求，遍历，然后验证是否有权限，这个权限是在配置文件里面配置的，详见http://blog.51cto.com/13579730/2072467
如果没有权限，返回SUBACK报文中标记该订阅状态为失败，如果有权限，检查topic是否有效如果有效，获取qos，如果无效标记为失败。
校验之后得到一个List<MqttTopicSubscription>，再根据这个list生成SUBACK

4.将RunningSubscription的状态从VERIFIED修改成STORED，这里面用到了ConcurrentHashMap.replace(key,oldvalue,newvlaue)这个原子操作，如果修改失败表面，这个订阅请求已经存在。
5.开始存储订阅请求，这里存储订阅请求

private List<Subscription> doStoreSubscription(List<MqttTopicSubscription> ackTopics, String clientID) {
    ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);

    List<Subscription> newSubscriptions = new ArrayList<>();
    for (MqttTopicSubscription req : ackTopics) {
        // TODO this is SUPER UGLY
        if (req.qualityOfService() == FAILURE) {
            continue;
        }
        Subscription newSubscription =
                new Subscription(clientID, new Topic(req.topicName()), req.qualityOfService());

        clientSession.subscribe(newSubscription);//存储到用户的session里面，用以表明该client订阅了哪些请求
        newSubscriptions.add(newSubscription);
    }
    return newSubscriptions;
}

    我们先看存储到用户的session这一步
    public boolean subscribe(Subscription newSubscription) {
    LOG.info("Adding new subscription. ClientId={}, topics={}, qos={}", newSubscription.getClientId(),
        newSubscription.getTopicFilter(), newSubscription.getRequestedQos());
    boolean validTopic = newSubscription.getTopicFilter().isValid();
    if (!validTopic) {
        LOG.warn("The topic filter is not valid. ClientId={}, topics={}", newSubscription.getClientId(),
            newSubscription.getTopicFilter());
        // send SUBACK with 0x80 for this topic filter
        return false;
    }
    ClientTopicCouple matchingCouple = new ClientTopicCouple(this.clientID, newSubscription.getTopicFilter());
    Subscription existingSub = subscriptionsStore.getSubscription(matchingCouple);
    // update the selected subscriptions if not present or if has a greater qos
    if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
        if (existingSub != null) {
            LOG.info("Subscription already existed with a lower QoS value. It will be updated. ClientId={}, " +
                "topics={}, existingQos={}, newQos={}", newSubscription.getClientId(),
                newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos());
            subscriptions.remove(newSubscription);
        }
        subscriptions.add(newSubscription);//存储到内存的session
        subscriptionsStore.addNewSubscription(newSubscription);//存储到别的地方
    }
    return true;
}
            这里面先创建了一个ClientTopicCouple对，然后从订阅集合里面查询是否已经存在这个订阅，如果不存在或者新的订阅的qos要高于就的订阅的qos，则会把订阅添加到订阅集合里面，这里有两个存储，一个是Set<Subscription>，一个是Map<Topic, Subscription> subscriptions（这个在ISessionsStore的具体实现里面）moquette在这里面做了冗余，即内存里面会存一分，同时允许通过ISessionsStore存储到外部。

6.我们接着看processSubscribe，这个方法会返回一个新的list
接着会遍历这个返回的list,存储到SubscriptionsDirectory里面，这个维护所有的client直接的发布订阅关系，是moquette里面一个非常重要的对象了，里面维护者一颗topic树，这个后面单独讲
7.发送SUBACK
8.发布retain消息，这里面也讲解一下，这一步的作用在于，如果一个client发布了新的订阅，那么必须遍历那些retain消息，如果这些新的订阅，确实能够匹配这些retain消息，必须将这些retain消息发送给他们。//这里moquette的处理是遍历map，这样的话，当retain消息特别大的时候，效率是非常低的，会很容易拖垮那些对吞吐率和性能要求比较高的系统的。

    private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) {
    LOG.info("Retrieving retained messages CId={}, topics={}", newSubscription.getClientId(),
            newSubscription.getTopicFilter());

    // scans retained messages to be published to the new subscription
    // TODO this is ugly, it does a linear scan on potential big dataset
    Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {

        @Override
        public boolean match(Topic key) {
            return key.match(newSubscription.getTopicFilter());
        }
    });

    if (!messages.isEmpty()) {
        LOG.info("Publishing retained messages CId={}, topics={}, messagesNo={}",
            newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size());
    }
    ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
    this.internalRepublisher.publishRetained(targetSession, messages);

    // notify the Observables
    m_interceptor.notifyTopicSubscribed(newSubscription, username);
}

另外，用以匹配订阅的topic与retain消息的topic是否匹配的方法也非常不完善。具体的原因大家可以看一下这里
io.moquette.spi.impl.subscriptions.Topic#match

9.从ConcurrentMap<RunningSubscription, SubscriptionState>移除该订阅请求。

整个RunningSubscription的状态会从VERIFIED到STORED，这代表了整个处理过程的最重要的两个步骤。
下一篇会讲解moquette对PUBLISH报文的处理