这期内容当中小编将会给大家带来有关如何理解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，这里面由于涉及到比较大的计算，所以单独讲解
