一分钟在SpringBoot项目中使用EMQ

先展示最终的结果:

生产者端:

@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 final Map<String, MsgHandler> handlerMap = new HashMap<>(); @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); } }

控制台结果展示:

---

补充JsonUtil

public class JsonUtil { /** * 从json字符串中根据nodeName获取值 * @param nodeName * @param json * @return * @throws IOException */ public static String getValueByNodeName(String nodeName, String json) throws IOException { ObjectMapper objectMapper = new ObjectMapper(); JsonNode jsonNode = objectMapper.readTree(json); JsonNode node = jsonNode.findPath(nodeName); if(node == null) return null; return node.asText(); } /** * 根据nodeName获取节点内容 * @param nodeName * @param json * @return * @throws IOException */ public static JsonNode getNodeByName(String nodeName, String json) throws IOException { ObjectMapper objectMapper = new ObjectMapper(); return objectMapper.readTree(json).findPath(nodeName); } /** * 反序列化 * @param json * @param clazz * @param <T> * @return * @throws IOException */ public static <T> T getByJson(String json, Class<T> clazz) throws IOException { ObjectMapper mapper = new ObjectMapper(); // 在反序列化时忽略在 json 中存在但 Java 对象不存在的属性 mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); // 在序列化时日期格式默认为 yyyy-MM-dd'T'HH:mm:ss.SSSZ mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false); return mapper.readValue(json, clazz); } /** * 反序列化（驼峰转换） * @param json * @param clazz * @param <T> * @return * @throws IOException */ public static <T> T getByJsonSNAKE(String json, Class<T> clazz) throws IOException { ObjectMapper mapper = new ObjectMapper(); // 在反序列化时忽略在 json 中存在但 Java 对象不存在的属性 mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); // 在序列化时日期格式默认为 yyyy-MM-dd'T'HH:mm:ss.SSSZ mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false); // 设置驼峰和下划线之间的映射 mapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE); return mapper.readValue(json, clazz); } /** * 序列化 * @param object * @return * @throws JsonProcessingException */ public static String serialize(Object object) throws JsonProcessingException { ObjectMapper mapper = new ObjectMapper(); return mapper.writeValueAsString(object); } /** * 序列化（驼峰转换） * @param object * @return * @throws JsonProcessingException */ public static String serializeSNAKE(Object object) throws JsonProcessingException { ObjectMapper mapper = new ObjectMapper(); // 设置驼峰和下划线之间的映射 mapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE); return mapper.writeValueAsString(object); } public static JsonNode getTreeNode(String json) throws JsonProcessingException { ObjectMapper objectMapper = new ObjectMapper(); return objectMapper.readTree(json); } /** * 将对象转map * @param obj * @return * @throws IOException */ public static Map<String,Object> convertToMap(Object obj) throws IOException { ObjectMapper mapper = new ObjectMapper(); return mapper.readValue(serialize(obj),Map.class); } }