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

这一篇开始讲解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报文的处理

原文地址：http://blog.51cto.com/13579730/2073914

时间： 2024-07-31 21:42:37

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

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

先上一个图,大概说明一下moquette 的类之间的关系 一.ProtocolProcessor类该类是moquette里面的最终要的类,负责所有报文的处理,持有所有各模块功能的实现对象的引用, 下面详细介绍 protected ConnectionDescriptorStore connectionDescriptors;//所有的连接描述符文存储,即clientId与通道之间的映射集合 protected ConcurrentMap<RunningSubscription, Subscrip

mqtt协议-broker之moqutte源码研究三之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); LO

mqtt协议-broker之moqutte源码研究一

mqtt协议的broker有很多,但是java的支持集群的并不多,之前调研过一番,发现moqutte基本满足需求,就想着基于这个在自己做二次开发.后面会逐渐把自己对moqutte的研究发布出来,希望能给有相同需求的同学一定的参考意义.github地址:https://github.com/andsel/moquette一.将代码倒入idea找到启动类启动报错,是因为找不到moquette的配置文件跟踪源码moquette的配置文件地址是config/moquette.conf因为咱们是直接启动的

mqtt协议-broker之moqutte源码研究五之UNSUBSCRIBE与DISCONN报文处理

本文讲解moquette对UNSUBSCRIBE和DISCONNECT的处理 先说UNSUBSCRIBE,代码比较简单 public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { List<String> topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); LOG.info("Pr

mqtt协议-broker之moqutte源码研究六之集群

moquette的集群功能是通过Hazelcast来实现的,对Hazelcast不了解的同学可以自行Google以下.在讲解moquette的集群功能之前需要讲解一下moquette的拦截器,因为moquette对Hazelcast的集成本身就是通过拦截器来实现的. 一.拦截器io.moquette.spi.impl.ProtocolProcessor类里面有一个BrokerInterceptor类,这个类就是broker拦截器,这个对象,在processConnect,processPubAc

Chrome自带恐龙小游戏的源码研究(二)

在上一篇<Chrome自带恐龙小游戏的源码研究(一)>中实现了地面的绘制和运动,这一篇主要研究云朵的绘制. 云朵的绘制通过Cloud构造函数完成.Cloud实现代码如下: 1 Cloud.config = { 2 HEIGHT:14, //云朵sprite的高度 3 MAX_CLOUD_GAP:400, //两朵云之间的最大间隙 4 MAX_SKY_LEVEL:30, //云朵的最大高度 5 MIN_CLOUD_GAP:100, //两朵云之间的最小间隙 6 MIN_SKY_LEVEL:71,

Android Launcher源码研究(二) 加载app流程1

今天主要分析Android Launcher源码中的一些重要类之间的关系,基本的加载流程.先来看一个类图 Launcher.java 是主Activity 在onCreate方法里面初始化了LauncherMode实例. LauncherApplication app = ((LauncherApplication)getApplication()); mModel = app.setLauncher(this); 直接进入LauncherApplication.java的方法 Launcher

Nginx源码研究二：NGINX的网络IO

NGINX作为服务端的应用程序,在客户端发出数据后,服务端在做着这样一些处理,数据先会经过网卡,网卡会和操作系统做交互,经过操作系统的协议栈处理,再和不同的应用程序交互. 在这里面涉及两个概念,一个是用户态,一个是内核态.应用程序通过系统调用函数进入内核空间,内核运行进行数据准备和数据拷贝等工作.对于NGINX来说,他是作为应用程序和操作系统交互,即是用户态和内核态的之间的交互,NGINX和内核交互方式有很多,例如open(),read() 等都是在和内核交互,而对于网络IO来说,我们知道lin

Nginx源码研究二：NGINX的内存管理

关于nginx的内存使用,我们先看代码,下面是nginx_cycle.c中对全局数据结构cycle的初始化过程 pool = ngx_create_pool(NGX_CYCLE_POOL_SIZE, log); //申请16K的内存池 if (pool == NULL) { return NULL; } pool->log = log; cycle = ngx_pcalloc(pool, sizeof(ngx_cycle_t)); if (cycle == NULL) { ngx_destroy