Mqtt消费端实现的几种方式

2024-09-03 22:36
文章标签 实现 方式 几种 mqtt 消费

本文主要是介绍Mqtt消费端实现的几种方式，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

此处测试的mqtt的Broker是使用的EMQX 5.7.1，可移步至https://blog.csdn.net/tiantang_1986/article/details/140443513查看详细介绍

一、方式1

添加必要的依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId>
</dependency>

配置

# mqtt 服务端配置
spring:# mqtt 配置mqtt:url: tcp://127.0.0.1:1883,tcp://127.0.0.2:1883clientId: "00000001"       # 客户端Id（不可重复）username: <访问用户名>      # 认证的用户名password: <访问密码>        # 认证的密码qos: 1topic: test/#              # 监听的topic

读取配置文件

import org.apache.commons.lang3.StringUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;@Data
@Configuration
@ConfigurationProperties(prefix = "spring.mqtt")
public class MqttConfig {private String username;private String password;private String url;private String clientId;private String topic;private Integer qos;@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();MqttConnectOptions options = new MqttConnectOptions();options.setUserName(username);options.setPassword(password.toCharArray());if (StringUtils.isNotBlank(url) && url.contains(",")) {options.setServerURIs(url.split(","));} else {options.setServerURIs(new String[]{url});}        options.setCleanSession(true);//自动重连options.setAutomaticReconnect(true);//设置超时时间，单位为秒options.setConnectionTimeout(0);//设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线options.setKeepAliveInterval(90);//设置遗嘱消息options.setWill("will_topic", (this.clientId + "与服务器断开连接").getBytes(), qos, false);factory.setConnectionOptions(options);factory.setPersistence(new MemoryPersistence());return factory;}
}

MQTT消息入站配置

@Slf4j
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {@Resourceprivate MqttConfig mqttConfig;@Resourceprivate MqttPahoClientFactory mqttClientFactory;@Resourceprivate MqttMessageReceiver mqttMessageReceiver;@Beanpublic MessageChannel mqttInBoundChannel() {return new PublishSubscribeChannel();}@Beanpublic MessageProducerSupport mqttInbound() {MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientFactory, mqttConfig.getTopic());DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();//传输Hex数据，如果是String则可使用默认值falseconverter.setPayloadAsBytes(true);adapter.setConverter(converter);adapter.setRecoveryInterval(10000);adapter.setQos(mqttConfig.getQos());adapter.setOutputChannel(mqttInBoundChannel());return adapter;}@Bean@ServiceActivator(inputChannel = "mqttInBoundChannel")public MessageHandler mqttMessageHandler() {return this.mqttMessageReceiver;}
}

消费者

@Slf4j
@Component
public class MqttMessageReceiver implements MessageHandler {@Resourceprivate DataConvertStrategyFactory convertStrategyContext;@Overridepublic void handleMessage(Message<?> message) throws MessagingException {MessageHeaders headers = message.getHeaders();String topic = (String) headers.get(MqttHeaders.RECEIVED_TOPIC);if (StringUtils.isNotBlank(topic)) {return;}byte[] payload = (byte[]) message.getPayload();log.info("topic: {}, message: {}", topic, HexUtils.bytesToHex(payload));//从topic中获取clientId，topic的格式：{业务}/{clientId}/{事件标识}Map<String, String> map = MqttDataConverter.covertTopic(topic);String clientId = map.get("clientId");log.info("clientId: {}", clientId);//topic中的事件标识String eventUrl = map.get("event");//自定义的enum，主要用来消息处理消息分组，相同组可以使用相同的数据转换服务Event[] events = Event.values();String deviceId = clientId;Arrays.stream(events).filter(item -> item.getEvent().equals(eventUrl)).findFirst().ifPresent(item -> {//使用策略模式实现DataConvertService convertService = convertStrategyContext.getStrategy(item.getGroup());convertService.convert(deviceId, eventUrl, payload);});}
}

数据转换服务接口，具体的数据解析只要实现这个接口就行

public interface DataConvertService {/*** 转换数据** @param clientId 设备SN* @param topic  topic* @param data  数据* @return*/Boolean convert(String clientId, String topic, byte[] data);/*** 获取转换器** @return*/String getConverter();
}

MQTT数据转换策略工厂

@Component
public class DataConvertStrategyFactory implements InitializingBean {@Resourceprivate List<DataConvertService> handlers;private Map<String, DataConvertService> dataConvertServiceMap = new ConcurrentHashMap<>();/*** 初始化*/@Overridepublic void afterPropertiesSet() {//进行初始化if (CollectionUtils.isNotEmpty(handlers)) {handlers.forEach(item -> {dataConvertServiceMap.put(item.getConverter(), item);});}}/*** 返回实际处理对象** @param strategy 处理策略* @return 实际处理对象*/public DataConvertService getStrategy(String strategy) {return dataConvertServiceMap.get(strategy);}
}

二、方式2

使用EMQX的Webhook钩子
首先创建钩子函数，把需要监听的事件加上处理逻辑，示例：

@Slf4j
@RequestMapping("/mqtt/client")
@RestController
public class ClientController {@PostMapping("/webhook")public Result webhook(@RequestBody Map<String, Object> message) {log.info("webhook map：{}", message);String action = (String) message.get("action");String clientid = (String) message.get("clientid");if ("client_connected".equals(action)) {log.info("client:{} 上线", clientid);}if ("client_disconnected".equals(action)) {log.info("client:{} 下线", clientid);}if ("message.publish".equals(action)) {log.info("已接收到 client:{} 的消息：{}", clientid, message.get("payload"));}return Result.success("OK");}
}

然后在EMQX的Dashboard中创建Webhook，可以选择多个触发器
在这里插入图片描述
填好URL后可以进行测试，之后使用MQTTX进行消息发送测试
在这里插入图片描述
控制台输出日志
在这里插入图片描述

三、方式3

package com.iinplus.mqtt.handler;import com.iinplus.mqtt.config.MqttConfig;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;import javax.annotation.Resource;@Slf4j
@Component
public class MqttSubscriber implements InitializingBean {@Resourceprivate MqttConfig config;@Overridepublic void afterPropertiesSet() {try {MqttClient client = new MqttClient(config.getUrl(), config.getClientId(), new MemoryPersistence());MqttConnectOptions options = new MqttConnectOptions();options.setUserName(config.getUsername());options.setPassword(config.getPassword().toCharArray());options.setCleanSession(true);options.setAutomaticReconnect(true);options.setConnectionTimeout(0);client.connect(options);client.subscribe(config.getTopic());//设置消息回调client.setCallback(new MqttMsgHandler());} catch (MqttException e) {log.error("MqttException：", e);}}
}

消息回调处理

@Slf4j
public class MqttMsgHandler implements MqttCallback {@Overridepublic void connectionLost(Throwable t) {// 连接丢失log.info("Connection lost：", t);}@Overridepublic void messageArrived(String topic, MqttMessage message) {// 接收到消息log.info("Message arrived：" + new String(message.getPayload()));}@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {// 消息发送成功log.info("Delivery complete");}
}

这篇关于Mqtt消费端实现的几种方式的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1134238。

相关文章

Linux系统中查询JDK安装目录的几种常用方法

Linux系统中查询JDK安装目录的几种常用方法

《Linux系统中查询JDK安装目录的几种常用方法》：本文主要介绍Linux系统中查询JDK安装目录的几种常用方法,方法分别是通过update-alternatives、Java命令、环境变量及目... 目录方法 1：通过update-alternatives查询（推荐）方法 2：检查所有已安装的 JDK方
阅读更多...
RabbitMQ消费端单线程与多线程案例讲解

RabbitMQ消费端单线程与多线程案例讲解

《RabbitMQ消费端单线程与多线程案例讲解》文章解析RabbitMQ消费端单线程与多线程处理机制,说明concurrency控制消费者数量,max-concurrency控制最大线程数,prefe... 目录 一、基础概念详细解释：举个例子：✅ 单消费者 + 单线程消费❌ 单消费者 + 多线程消费❌ 多
阅读更多...
Java获取当前时间String类型和Date类型方式

Java获取当前时间String类型和Date类型方式

《Java获取当前时间String类型和Date类型方式》：本文主要介绍Java获取当前时间String类型和Date类型方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录Java获取当前时间String和Date类型String类型和Date类型输出结果总结Java获取
阅读更多...
Python实现批量提取BLF文件时间戳

Python实现批量提取BLF文件时间戳

《Python实现批量提取BLF文件时间戳》BLF（BinaryLoggingFormat）作为Vector公司推出的CAN总线数据记录格式,被广泛用于存储车辆通信数据,本文将使用Python轻松提取... 目录一、为什么需要批量处理 BLF 文件二、核心代码解析：从文件遍历到数据导出1. 环境准备与依赖库
阅读更多...
linux下shell脚本启动jar包实现过程

linux下shell脚本启动jar包实现过程

《linux下shell脚本启动jar包实现过程》确保APP_NAME和LOG_FILE位于目录内,首次启动前需手动创建log文件夹,否则报错,此为个人经验,供参考,欢迎支持脚本之家... 目录linux下shell脚本启动jar包样例1样例2总结linux下shell脚本启动jar包样例1#!/bin
阅读更多...
go动态限制并发数量的实现示例

go动态限制并发数量的实现示例

《go动态限制并发数量的实现示例》本文主要介绍了Go并发控制方法,通过带缓冲通道和第三方库实现并发数量限制,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面... 目录带有缓冲大小的通道使用第三方库其他控制并发的方法因为go从语言层面支持并发，所以面试百分百会问到
阅读更多...
Go语言并发之通知退出机制的实现

Go语言并发之通知退出机制的实现

《Go语言并发之通知退出机制的实现》本文主要介绍了Go语言并发之通知退出机制的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录1、通知退出机制1.1 进程/main函数退出1.2 通过channel退出1.3 通过cont
阅读更多...
Python实现PDF按页分割的技术指南

Python实现PDF按页分割的技术指南

《Python实现PDF按页分割的技术指南》PDF文件处理是日常工作中的常见需求,特别是当我们需要将大型PDF文档拆分为多个部分时,下面我们就来看看如何使用Python创建一个灵活的PDF分割工具吧... 目录需求分析技术方案工具选择安装依赖完整代码实现使用说明基本用法示例命令输出示例技术亮点实际应用场景扩
阅读更多...
C#监听txt文档获取新数据方式

C#监听txt文档获取新数据方式

《C#监听txt文档获取新数据方式》文章介绍通过监听txt文件获取最新数据,并实现开机自启动、禁用窗口关闭按钮、阻止Ctrl+C中断及防止程序退出等功能,代码整合于主函数中,供参考学习... 目录前言一、监听txt文档增加数据二、其他功能1. 设置开机自启动2. 禁止控制台窗口关闭按钮3. 阻止Ctrl +
阅读更多...
java如何实现高并发场景下三级缓存的数据一致性

java如何实现高并发场景下三级缓存的数据一致性

《java如何实现高并发场景下三级缓存的数据一致性》这篇文章主要为大家详细介绍了java如何实现高并发场景下三级缓存的数据一致性,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 下面代码是一个使用Java和Redisson实现的三级缓存服务，主要功能包括：1.缓存结构：本地缓存：使
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2