先展示最终的结果:
生产者端:
@RestController
@RequiredArgsConstructor
public class TestController {
private final MqttProducer mqttProducer;
@GetMapping("/test")
public String test() {
User build = User.builder().age(100).sex(1).address("世界潍坊渤海之眼").build();
// 延时发布
mqttProducer.send("$delayed/10/cookie", 2, JSON.toJSONString(build));
return "ok";
}
}
消费者端
/**
* @author : Cookie
* date : 2024/1/30
*/
@Component
@Topic("cookie")
public class TestTopicHandler implements MsgHandler {
@Override
public void process(String jsonMsg) {
User user = JSON.parseObject(jsonMsg, User.class);
System.out.println(user);
}
}
控制台结果:
具体解释在之前的笔记中, 需要的话可以查看 EMQ的介绍及整合SpringBoot的使用-CSDN博客
OK, 下面我们就开始实现上面的逻辑, 你要做的就是把 1-9 复制到项目即可
1. 依赖导入
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
2. yml 配置
# 顶格
mqtt:
client:
username: admin
password: public
serverURI: tcp://192.168.200.128:1883
clientId: monitor.task.${
random.int[10000,99999]} # 注意: emq的客户端id 不能重复
keepAliveInterval: 10 #连接保持检查周期 秒
connectionTimeout: 30 #连接超时时间 秒
producer:
defaultQos: 2
defaultRetained: false
defaultTopic: topic/test1
consumer:
consumerTopics: $queue/cookie/#, $share/group1/yfs1024 #不带群组的共享订阅 多个主题逗号隔开
# $queue/cookie/#
# 以$queue开头，不带群组的共享订阅 多个客户端只能有一个消费者消费
# $share/group1/yfs1024
# 以$share开头，群组的共享订阅 多个客户端订阅
# 如果在一个组 只能有一个消费者消费
# 如果不在一个组 都可以消费
3. 属性配置
@Data
@Configuration
@ConfigurationProperties(prefix = "mqtt.client")
public class MqttConfigProperties {
private int defaultProducerQos;
private boolean defaultRetained;
private String defaultTopic;
private String username;
private String password;
private String serverURI;
private String clientId;
private int keepAliveInterval;
private int connectionTimeout;
}
4. 定义主题注解
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({
ElementType.TYPE})
public @interface Topic {
String value();
}
5.Mqtt配置类
@Data
@Slf4j
@Configuration
@RequiredArgsConstructor
public class MqttConfig {
private final MqttConfigProperties configProperties;
private final MqttCallback mqttCallback;
@Bean
public MqttClient mqttClient() {
try {
MqttClient client = new MqttClient(configProperties.getServerURI(), configProperties.getClientId(), mqttClientPersistence());
client.setManualAcks(true); //设置手动消息接收确认
mqttCallback.setMqttClient(client);
client.setCallback(mqttCallback);
client.connect(mqttConnectOptions());
return client;
} catch (MqttException e) {
log.error("emq connect error", e);
return null;
}
}
@Bean
public MqttConnectOptions mqttConnectOptions() {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(configProperties.getUsername());
options.setPassword(configProperties.getPassword().toCharArray());
options.setAutomaticReconnect(true);//是否自动重新连接
options.setCleanSession(true);//是否清除之前的连接信息
options.setConnectionTimeout(configProperties.getConnectionTimeout());//连接超时时间
options.setKeepAliveInterval(configProperties.getKeepAliveInterval());//心跳
options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);//设置mqtt版本
return options;
}
public MqttClientPersistence mqttClientPersistence() {
return new MemoryPersistence();
}
}
6. 定义消息处理接口
/**
* 消息处理接口
*/
public interface MsgHandler {
void process(String jsonMsg) throws IOException;
}
7.定义消息上下文
/**
* 消息处理上下文, 通过主题拿到topic
*/
public interface MsgHandlerContext{
MsgHandler getMsgHandler(String topic);
}
8. 定义回调类
@Component
@Slf4j
public class MqttCallback implements MqttCallbackExtended {
// 需要订阅的topic配置
@Value("${mqtt.consumer.consumerTopics}")
private List<String> consumerTopics;
@Autowired
private MsgHandlerContext msgHandlerContext;
@Override
public void connectionLost(Throwable throwable) {
log.error("emq error.", throwable);
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
log.info("topic:" + topic + " message:" + new String(message.getPayload()));
//处理消息
String msgContent = new String(message.getPayload());
log.info("接收到消息:" + msgContent);
try {
// 根据主题名称 获取 该主题对应的处理器对象
// 多态 父类的引用指向子类的对象
MsgHandler msgHandler = msgHandlerContext.getMsgHandler(topic);
if (msgHandler == null) {
return;
}
msgHandler.process(msgContent); //执行
} catch (IOException e) {
log.error("process msg error,msg is: " + msgContent, e);
}
//处理成功后确认消息
mqttClient.messageArrivedComplete(message.getId(), message.getQos());
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
}
@Override
public void connectComplete(boolean b, String s) {
log.info("连接成功");
//和EMQ连接成功后根据配置自动订阅topic
if (consumerTopics != null && consumerTopics.size() > 0) {
// 循环遍历当前项目中配置的所有的主题.
consumerTopics.forEach(t -> {
try {
log.info(">>>>>>>>>>>>>>subscribe topic:" + t);
// 订阅当前集群中所有的主题 消息服务质量 2 -> 至少收到一个
mqttClient.subscribe(t, 2);
} catch (MqttException e) {
log.error("emq connect error", e);
}
});
}
}
private MqttClient mqttClient;
// 在配置类中调用传入连接
public void setMqttClient(MqttClient mqttClient) {
this.mqttClient = mqttClient;
}
}
8. 消息处理类加载器
作用: 将Topic跟处理类对应 通过 handlerMap
/**
* 消息处理类加载器
* 作用:
* 1. 因为实现了Spring 的 ApplicationContextAware 接口, 项目启动后就会运行实现的方法
* 2. 获取MsgHandler接口的所有的实现类
* 3. 将实现类上的Topic注解的值,作为handlerMap的键,实现类(处理器)作为对应的值
*/
@Component
public class MsgHandlerContextImp implements ApplicationContextAware, MsgHandlerContext {
private Map<String, MsgHandler> handlerMap = Maps.newHashMap();
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
// 从spring容器中获取 <所有> 实现了MsgHandler接口的对象
// key 默认类名首字母小写 value 当前对象
Map<String, MsgHandler> map = applicationContext.getBeansOfType(MsgHandler.class);
map.values().forEach(obj -> {
// 通过反射拿到注解中的值 即 当前类处理的 topic
String topic = obj.getClass().getAnnotation(Topic.class).value();
// 将主题和当前主题的处理类建立映射
handlerMap.put(topic,obj);
});
}
@Override
public MsgHandler getMsgHandler(String topic) {
return handlerMap.get(topic);
}
}
9. 封装消息生产者
@Slf4j
@Component
public class MqttProducer {
// @Value() 读取配置 当然也可以批量读取配置,这里就一个一个了
@Value("${mqtt.producer.defaultQos}")
private int defaultProducerQos;
@Value("${mqtt.producer.defaultRetained}")
private boolean defaultRetained;
@Value("${mqtt.producer.defaultTopic}")
private String defaultTopic;
@Autowired
private MqttClient mqttClient;
public void send(String payload) {
this.send(defaultTopic, payload);
}
public void send(String topic, String payload) {
this.send(topic, defaultProducerQos, payload);
}
public void send(String topic, int qos, String payload) {
this.send(topic, qos, defaultRetained, payload);
}
public void send(String topic, int qos, boolean retained, String payload) {
try {
mqttClient.publish(topic, payload.getBytes(), qos, retained);
} catch (MqttException e) {
log.error("publish msg error.",e);
}
}
public <T> void send(String topic, int qos, T msg) throws JsonProcessingException {
String payload = JsonUtil.serialize(msg);
this.send(topic,qos,payload);
}
}
最终的实现的结果
- 生产者端: 在需要发送消息的地方注入 MqttProducer 发送消息
- 消费者端: 在需要处理对应主题的类上 实现 MsgHandler接口
代码示例
生产者端
@RestController
@RequiredArgsConstructor
public class TestController {
private final MqttProducer mqttProducer;
@GetMapping("/test")
public String test() {
User build = User.builder().age(100).sex(1).address("世界潍坊渤海之眼").build();
// 延时发布
mqttProducer.send("$delayed/10/cookie", 2, JSON.toJSONString(build));
return "ok";
}
}
消费者端
@Component
@Topic("cookie")
public class TestTopicHandler implements MsgHandler {
@Override
public void process(String jsonMsg) {
User user = JSON.parseObject(jsonMsg, User.class);
System.out.println(user);
}
}
控制台结果展示:
