springboot+rabbitmq搭建mqtt协议实现订阅发布（亲测9w消息并发）

一、mqtt协议简单介绍mqtt是一种基于发布/订阅publish/subscribe模式的轻量级通讯协议通过订阅相应的主题来获取消息是物联网Internetof Thing中的一个标准传输协议。二、rabbitmq的安装部署1. 彻底卸载旧版本 Erlang# 1. 查看所有已安装的 Erlang 相关包 rpm -qa | grep erlang # 2. 批量卸载所有 Erlang 包无需手动改包名命令自动匹配 sudo rpm -e --nodeps $(rpm -qa | grep erlang) # 3. 强制删除残留文件防止卸载不彻底 sudo rm -rf /usr/lib64/erlang /usr/share/erlang #4.再次执行 rpm -qa | grep erlang无任何输出即卸载干净。 rpm -qa | grep erlang2. 下载Erlang rpm 安装包和rabbitmq rpm安装包RPM 下载包版本地址https://packagecloud.io/rabbitmq/erlang下载RabbitMQ rpm 安装包 https://github.com/rabbitmq/rabbitmq-server/releases/*注意版本统一安装erlangrpm -ivh erlang-23.2.7-1.el7.x86_64.rpmerl -v安装rabbitmqyum install -y socatrpm -ivh rabbitmq-server-3.9.15-1.el7.noarch.rpm启动rabbitmqsystemctl start rabbitmq-server查看rabbitmq状态systemctl status rabbitmq-server设置开机自启重启服务器后自动运行sudo systemctl enable rabbitmq-server3、添加用户创建管理员用户添加root管理员用户取代guest用户**rabbitmqctl add_user root rootrabbitmqctl set_user_tags root administratorrabbitmqctl set_permissions -p / root .* .* .*删除guestrabbitmqctl delete_user guest查看 root 用户的 tags 为 [administrator] 即成功rabbitmqctl list_users创建普通用户并设置权限仅用于发送订阅消息创建v-hostrabbitmqctl add_vhost /uav创建用户rabbitmqctl add_user uavadmin uavadmin设置用户角色,无法登陆管理控制台通常就是普通的生产者和消费者。rabbitmqctl set_user_tags uavadmin none设置用户在uav 下的权限rabbitmqctl set_permissions -p /uav uavadmin .* .* .*设置用户在默认/” 下的权限rabbitmqctl set_permissions -p / uavadmin .* .* .*设置主题权限可订阅和发布消息rabbitmqctl set_topic_permissions -p /uav uavadmin amq.topic .* .*4、常用命令后续管理用重启 RabbitMQ : systemctl restart rabbitmq-server停止 RabbitMQ : systemctl stop rabbitmq-server查看 RabbitMQ 日志 : tail -f /var/log/rabbitmq/rabbitlocalhost.log查看所有队列 : rabbitmqctl list_queues修改用户密码 : rabbitmqctl change_password admin 新密码删除匿名用户安全 : rabbitmqctl delete_user guest三、启用 rabbitmq的mqtt协议和RabbitMQWeb管理界面rabbitmq插件启用启动RabbitMQWeb管理界面rabbitmq-plugins enable rabbitmq_management启用 rabbitmq的mqtt协议rabbitmq-plugins enable rabbitmq_mqtt启用 rabbitmq的web_mqtt协议(不使用js订阅发布可以不启用)rabbitmq-plugins enable rabbitmq_web_mqtt查看插件状态 E显式启用 e隐式启用rabbitmq-plugins list开放外网访问并重启防火墙firewall-cmd --zonepublic --add-port5672/tcp --permanentfirewall-cmd --zonepublic --add-port15672/tcp --permanentfirewall-cmd --zonepublic --add-port1883/tcp --permanentfirewall-cmd --zonepublic --add-port15675/tcp --permanent如果搭建rabbitmq集群模式需要把下面这个两个端口放开firewall-cmd --zonepublic --add-port4369/tcp --permanentfirewall-cmd --zonepublic --add-port25672/tcp --permanent重启防火墙systemctl restart firewalldfirewall-cmd --zonepublic --list-ports以上部署操作全部设定完毕重启rabbitmq服务使用创建root用户登录进入rabbitmq控制台*至此rabbitmq搭建mqtt安装部署结束下面进入代码实现环节四、代码实现先在pom中添加依赖包dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependencyapplication.yml的所需的配置mqtt: #MQTT-用户名 root username: uavadmin #MQTT-密码,需要解密 root password: uavadmin #MQTT-服务器连接地址如果有多个用逗号隔开如tcp://xxx.xxx.xx.xxx:1883,tcp://xxx.xxx.xxx.xxx:1883 hostUrl: tcp://192.168.2.128:1883,tcp://192.168.2.129:1883 #两个客户端的clientId不能相同生产者和消费者的clientId不能相同 pubClientId: pub-client-id-al68pq1w-dev subClientId: sub-client-id-9v83pp7c-dev #发布的主题--MQTT-默认的消息推送主题实际可在调用接口时指定 pubTopic: defaultTopic #订阅的主题 subTopic: gps-topic,oil-topic,broadcast-topic,fault-topic completionTimeout: 3000mqtt服务类package com.zdft.bhdcm.config.mtqq; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.messaging.*; import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; /** * mqtt服务类 * 一种基于发布/订阅publish/subscribe模式的轻量级通讯协议通过订阅相应的主题来获取消息 * 是物联网Internet of Thing中的一个标准传输协议 * ClientId是MQTT客户端的标识。MQTT服务端用该标识来识别客户端。因此ClientId必须是独立的。 * clientID需为全局唯一。如果不同的设备使用相同的clientID同时连接物联网平台那么先连接的那个设备会被强制断开。 */ Configuration IntegrationComponentScan Slf4j public class MqttServerConfig { Value(${mqtt.username}) private String username; Value(${mqtt.password}) private String password; Value(${mqtt.hostUrl}) private String hostUrl; Value(${mqtt.pubClientId}) private String pubClientId; Value(${mqtt.subClientId}) private String subClientId; Value(${mqtt.pubTopic}) private String pubTopic; Value(${mqtt.subTopic}) private String subTopic; Value(${mqtt.completionTimeout}) private int completionTimeout; /*factory*/ /** * mqtt客户工厂 * return */ Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions mqttConnectOptions new MqttConnectOptions(); mqttConnectOptions.setUserName(username); mqttConnectOptions.setPassword(password.toCharArray()); // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录 // 把配置里的 cleanSession 设为false客户端掉线后 服务器端不会清除session // 当重连后可以接收之前订阅主题的消息。当客户端上线后会接受到它离线的这段时间的消息 mqttConnectOptions.setCleanSession(false); mqttConnectOptions.setServerURIs(hostUrl.split(,)); // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线但这个方法并没有重连的机制 mqttConnectOptions.setKeepAliveInterval(20); mqttConnectOptions.setMaxInflight(1000); factory.setConnectionOptions(mqttConnectOptions); return factory; } /*sent*/ /** * mqtt出站通道 * return */ Bean public MessageChannel mqttOutboundChannel() { return new DirectChannel(); } /** * mqtt出站handler * * return {link MessageHandler} */ Bean ServiceActivator(inputChannel mqttOutboundChannel) public MessageHandler mqttOutboundHandler() { //MqttPahoMessageHandler初始化 MqttPahoMessageHandler handler new MqttPahoMessageHandler(pubClientId_send_handler_, mqttClientFactory()); //设置默认的qos级别 handler.setDefaultQos(1); //保留标志的默认值。如果没有mqtt_retained找到标题则使用它。如果提供了自定义则不使用它converter。这里不启用 handler.setDefaultRetained(false); //设置发布的主题 handler.setDefaultTopic(pubTopic); //当 时true调用者不会阻塞。相反它在发送消息时等待传递确认。默认值为false在确认交付之前发送阻止。 handler.setAsync(false); //当 async 和 async-events 都为 true 时会发出 MqttMessageSentEvent请参阅事件。它包含消息、主题、客户端库生成的messageId、clientId和clientInstance每次连接客户端时递增。当客户端库确认交付时会发出 MqttMessageDeliveredEvent。它包含 messageId、clientId 和 clientInstance使传递与发送相关联。任何 ApplicationListener 或事件入站通道适配器都可以接收这些事件。请注意有可能在 MqttMessageSentEvent 之前接收到 MqttMessageDeliveredEvent。默认值为false。 handler.setAsyncEvents(false); return handler; } /*receive*/ /** * mqtt输入通道 * return */ Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } /** * 入站 * return */ Bean public MessageProducer inbound() { //配置订阅端MqttPahoMessageDrivenChannelAdapter MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(subClientId_receive_inbound_, mqttClientFactory(), subTopic.split(,)); //设置超时时间 adapter.setCompletionTimeout(completionTimeout); //设置默认的消息转换类 adapter.setConverter(new DefaultPahoMessageConverter()); //设置qos级别 adapter.setQos(1); //设置入站管道 adapter.setOutputChannel(mqttInputChannel()); adapter.setTaskScheduler(new ConcurrentTaskScheduler()); return adapter; } /** * 消息处理程序 * return */ Bean ServiceActivator(inputChannel mqttInputChannel) public MessageHandler handler() { return new MessageHandler() { Override public void handleMessage(Message? message) throws MessagingException { MessageHeaders headers message.getHeaders(); log.info(headers: {}, headers); String topic headers.get(MqttHeaders.RECEIVED_TOPIC).toString(); log.info(订阅主题为: {}, topic); String[] topics subTopic.split(,); for (String t : topics) { if (t.equals(topic)) { log.info(订阅主题为:{};接收到该主题消息为:{},topic,message.getPayload().toString()); } } } }; } }mqtt网关发布端需要用到package com.zdft.bhdcm.config.mtqq;importorg.springframework.integration.annotation.MessagingGateway;importorg.springframework.integration.mqtt.support.MqttHeaders;importorg.springframework.messaging.handler.annotation.Header;importorg.springframework.stereotype.Component;/** *mqtt网关发布端需要用到 */Component MessagingGateway(defaultRequestChannelmqttOutboundChannel)publicinterface MqttGateway{/** * 发送到mqtt * * param payload 有效载荷 */voidsendToMqtt(String payload);/** * 发送到mqtt * * param topic 主题 * param payload 消息内容 */voidsendToMqtt(Header(MqttHeaders.TOPIC)String topic,String payload);/** * 发送到mqtt * * param topic 主题 * param qos qos * param payload 消息内容 */voidsendToMqtt(Header(MqttHeaders.TOPIC)String topic,Header(MqttHeaders.QOS)intqos,String payload);}测试发送和订阅五、mqtt.fx连接mqtt工具使用mqtt测试工具安装包https://pan.baidu.com/s/1oun7rMVJITOK9VSyO785HQ提取码l3cm1、配置连接及订阅配置mqtt连接配置用户名密码订阅gps-topic六、jmeter压测结果展示这里拟2w的消息并发量根据业务计算最高模拟测试9w消息并发量没出现问题如何使用移步到 https://blog.csdn.net/weixin_39393393/article/details/116640867?spm1001.2014.3001.55021、使用jmeter模拟2w并发量2、结果展示rabbitmq控制台展示后台打印mqtt订阅的消息jmeter压测报告