MQTT broker

retained消息

• pub端

  1 2	// 最后的参数retain：设置为true以保留消息, 可以修改该值做测试, 比如先pub然后停止，然后调用sub看看是否可以收到消息 ret = mosquitto_publish(mosq, NULL, TOPIC_NAME, str_json.size() + 1, str_json.c_str(), QOS, 1);

• 可以观察到
  1. broker只保留最后一个retain的消息
  2. 假如sub端写了session，那么broker会记录sub已经收到的消息，那么sub重启也不会收到retain的消息，因为retain的消息只发给新订阅的sub
  3. 假如sub端没有写session，或者重复订阅了主题，那么也会收到retain的消息

cleansession

• sub端

  1 2	static bool clean_session = false; mosq = mosquitto_new(CLIENT_ID, clean_session, NULL);

• 假设clean_session设置为false，也就是不清空会话，那么即使pub端发布的时候，sub不在线，sub上线后仍然会收到一定数量的消息。
• 这个session的历史会话，需要在broker的conf中进行配置，而mqtt的broker是默认没有配置这个内容的，所以是没有历史会话功能的。
• 需要注意的是，mqtt的broker是没有持久化的功能的，也就是当broker重启的时候，session的消息会消失；持久化功能需要接入emqx等云平台，并接入持久化系统。

遗嘱机制

• 遗嘱机制就是为了在pub端意外退出的时候，通知到所有订阅了该遗嘱主题的sub端，并携带一些遗嘱信息；

  1 2 3

  int qos = 1; bool retain = true; mosquitto_will_set(mosq, "topic/on/unexpected/disconnect", strlen("will message"), "will message", qos, retain);

• 值得注意的是，will retain的使用场景，假如订阅该遗嘱主题的客户端不能保证遗嘱消息发布的时候在线，那么需要将retain设置为true，确保订阅端上线的时候能够收到该主题的retained消息
• 意外断线：当broker没有收到disconnect报文的情况下，发现客户端心跳停止了，这时服务端就知道客户端时意外断线了
  1. 因网络故障或网络波动，设备在保持连接周期内未能通讯，连接被服务端关闭
  2. 设备意外掉电
  3. 设备尝试进行不被允许的操作而被服务端关闭连接，例如订阅自身权限以外的主题

接入EXQX云平台

• 优势：

1. 价格低廉
2. 功能齐全，支持持久化、集成kafka、用户身份安全认证方案和集成mosquitto brige（实现将已经使用mosquitto开源broker的方案直接使用接入云平台）等
3. 直接可以部署到阿里云、腾讯云等云平台，无需额外购买云服务器

发布订阅过滤器

• 当一个客户端订阅多个topic时，如何识别是从哪个主题订阅的消息？
  • 例如 同时订阅 mqtt/+ 和 mqtt/pm2.5 这两个主题的sub端，是没办法区分这两个主题的消息的。
• 解决：使用mqtt v5版本的设置和响应，通过identify来识别

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33

  // 初始化协议版本 mosqitto_int_option(mosq, MOSQ_OPT_PROTOCOL_VERSION, MQTT_PROTOCOL_V5); // callback设置 void mosquitto_message_callback_set(struct mosquitto *mosq, void (*on_message)(struct mosquitto *, void *, const struct mosquitto_message *)) { //... } void mosquitto_message_v5_callback_set(struct mosquitto *mosq, void (*on_message)(struct mosquitto *, void *, const struct mosquitto_message *, const mosquitto_property* props)) { //... } // 订阅设置 int mosquitto_subscribe(struct mosquitto *mosq, int *mid, const char *sub, int qos){ return mosquitto_subscribe_multiple(mosq, mid, 1, (char *const *const)&sub, qos, 0, NULL); } int mosquitto_subscribe_v5(struct mosquitto *mosq, int *mid, const char *sub, int qos, int options, const mosquitto_property *properties){ return mosquitto_subscribe_multiple(mosq, mid, 1, (char *const *const)&sub, qos, options, properties); } // property 结构体示例 struct mqtt5__property{ struct mqtt5__property *next; union{ uint8_t i8; uint16_t i16; uint32_t i32; uint32_t varint; struct mqtt__string bin; struct mqtt__string s; } value; struct mqtt__string name; int32_t identifier; // 这个标志位可以用来区别主题 bool client_generated; };

EMQX文档的重要性

1. 不论作为MQTT还是云平台接入的学习都很有帮助
2. 可以参考文档和网站中的商业应用案例理解mqtt

MQTT和Kafka有什么区别

• MQTT 和 Kafka 是完全不同的技术，虽然它们具有一些类似的特性，比如都基于发布订阅模式，都通过主题来路由消息等等。
• 但实际上，Kafka 更多地是专注于数据的存储和读取，针对实时性高的流式数据处理场景。而 MQTT 则侧重于海量的设备连接和主题路由，并且在不稳定的网络环境中进行实时的、可靠的消息交换。所以，MQTT 和 Kafka 的最大差异在于它们有着不同的适用场景，解决的是不同的问题。
• 在实际的生产应用中，MQTT 和 Kafka 经常会结合一起使用。MQTT 代理解决海量终端设备的接入问题，并将汇入的海量终端数据转发给 Kafka，再由 Kafka 分发给不同的业务应用程序，进行进一步的消息分析和处理。