JarCastingMQTT Java 客户端
1. mqtt相关介绍及文档
参考文档：
EMQ X: https://docs.emqx.io/tutorial/v3/cn/
EMQ X-共享订阅：https://docs.emqx.io/tutorial/v3/cn/advanced/share_subscribe.html
MQTT 一些概念： https://blog.csdn.net/u013151053/article/details/81366677
MQTT使用: https://www.cnblogs.com/sxkgeek/p/9140180.html
demo:
java：easy-mqtt/test目录 和 easy-mqtt-springboot-starter目录
js: mqtt-with-websocket.html
2. gradle 添加依赖
单独使用
implementation "com.github.giveme0101:easy-mqtt:${easymqtt.version}"
集成springboot
implementation "com.github.giveme0101:easy-mqtt-spring-boot-starter:${easymqttSpringbootStarter.version}"
3. 使用 - 集成 SpringBoot
1. application.yml 配置
mqtt:
# 生产者，发送端
producer:
# client-id不填动态生成
client-id: producer-1
server-uri: tcp://127.0.0.1:1883
username: admin
password: admin
# clean-session默认true，每次使用新session
clean-session: true
# qos默认值2，仅发送一次
qos: 2
# retained默认false，broker不暂存消息
retained: false
# 消费者，接收端
consumer:
# 固定client-id, clean-session=false 可以接受离线消息
# client-id不填动态生成
client-id: consumer-1
# clean-session默认true，每次使用新session
clean-session: false
server-uri: tcp://127.0.0.1:1883
username: admin
password: admin
topics:
- /test/#
- /topic/#
- /message/#
# 一个qos对应一个topic,数量必须相等。qos可以都不定义，不定义使用mqtt默认值：1
# qos取值 0,1,2
# qos = 0, 确保至多收到一次消息(只负责传送，发送过后就不管数据的传送情况)
# = 1, 确保至少收到一次消息
# = 2, 确保只收到一次消息 (使用两阶段提交2pc保证一执行)
qos:
- 2
- 2
- 2
2. Java 代码
producer:
1. config类添加
@EnableMqttProducer
2.
public xxxClass{
@Autowired
private MqttProducer producer;
...{
producer.publish("/topic/1", "message");
producer.publish("/topic/part/2", new HashMap());
producer.publish("/topic/3", new Object());
producer.publish("/cn/qd/news", new Object());
producer.publish("/cn/qd/weather", new HashMap());
}
}
consumer:
1. 如果配置1）和2）两种方式，则在topic满足条件的情况下都能收到消息
1）
a: 启动类添加@EnableMqttConsumer并将@MqttListener所在的包配置到scanBasePackages：
@EnableMqttConsumer(scanBasePackages = {"demo.listener"})
b: 类实现IMqttListener接口并添加@MqttListener并指定一个或多个topic（此topic必须在application.yml中注册），此topic可以模糊匹配，
如/topic/+/+ , 一个+可以匹配一级，可以匹配/topic/qd/1,不能匹配/topic/x
/topic/# , #必须放在结尾，匹配/tocpic/开头的所有topic
@MqttListener(topic = {"/topic/#", "/cn/qd/+"})
public class TopicListener implements IMqttListener {
@Override
publc void onMessage(String topic, MqttMessage message){
System.out.println("topic name: " + topic + "，message: " + message);
}
}
2）
a: 启动类添加@EnableMqttConsumer
b: 类实现IMqttMessageObserver接口并添加@Component等扫描注解，可以收到所有在application.yml中配置的topic消息
@Component
public class MqttMessageCenter implements IMqttMessageObserver {
@Override
public void onMessage(final String topic, final MqttMessage message) {
System.out.println("from MqttMessageCenter: " + topic + " - " + message);
}
}
4. 使用 -- 不集成SpringBoot
详见：
easy-mqtt测试用例， DynamicMqttClient.java
5. 共享订阅（基于EMQ X的实现）
创建共享订阅的两种方法：
主题前缀 实例
$queue/:topic sub $queue/up/data
$share/:group/:topic sub $share/group/up/data
共享订阅由三部分组成：
静态共享订阅标识符（$queue 与 $share）
组标识符（可选）
特定标准 MQTT 主 题（实际接收消息的主题）
$queue 和 $share 之间的差异：
$queue 之后的主题中所有消息将轮流发送到客户端，
$share 之后，您可以添加不同的组，例如:
$share/group_1/topic
$share/group_2/topic
$share/group_3/topic
当 EMQ X 向 topic 发送消息时，每个组都会收到该消息，并依次将其发送到该组中的设备。
共享订阅的使用:
共享订阅仅支持消费者，在yml中的mqtt.consumer.topics的每一个topic前添加 $queue/ 或 $share/group_name/， 注意包含结尾的‘/’, 如：
$share/group_1/cn/dq/# ==> cn/dq/#
$share/group_1//cn/dq/+ ==> /cn/dq/+
$queue//cn/dq/news ==> /cn/dq/news