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

一、mqtt协议简单介绍

mqtt是一种基于发布/订阅（publish/subscribe）模式的轻量级通讯协议，通过订阅相应的主题来获取消息，是物联网（Internet
of Thing）中的一个标准传输协议。

二、rabbitmq的安装部署

1. 安装Erlang环境

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

yum -y install ncurses-devel

2. 下载Erlang rpm 安装包和rabbitmq rpm安装包

rpm包自取：https://pan.baidu.com/s/1UGuxeEIYMK9hBHKYBClfTQ
提取码：tmfm

RPM 下载包版本地址：https://packagecloud.io/rabbitmq/erlang

下载RabbitMQ rpm 安装包： https://github.com/rabbitmq/rabbitmq-server/releases/

*注意版本统一

安装erlang

rpm -Uvh erlang-23.2.7-1.el7.x86_64.rpm

yum install -y erlang

erl -v

安装rabbitmq

yum install -y socat

rpm -Uvh rabbitmq-server-3.9.15-1.el7.noarch.rpm

yum install -y rabbitmq-server

启动rabbitmq

systemctl start rabbitmq-server

查看rabbitmq状态

systemctl status rabbitmq-server

3、添加用户

添加root用户取代guest用户

rabbitmqctl add_user root   root

rabbitmqctl set_user_tags root administrator

rabbitmqctl set_permissions -p "/" root ".*" ".*" ".*"

删除guest

rabbitmqctl delete_user guest  

创建普通用户并设置权限仅用于发送订阅消息
创建v-host

rabbitmqctl add_vhost /third_mqtt

创建用户

rabbitmqctl add_user third_client OP74X53Z

设置用户角色,无法登陆管理控制台，通常就是普通的生产者和消费者。

rabbitmqctl set_user_tags third_client none

设置用户在v-host下的权限

rabbitmqctl set_permissions -p /third_mqtt third_client ".*" ".*" ".*" 

设置用户在默认"/” v-host下的权限

rabbitmqctl set_permissions -p  /  third_client ".*" ".*" ".*" 

设置主题权限，可订阅和发布消息

rabbitmqctl set_topic_permissions -p /third_mqtt third_client amq.topic ".*" ".*"

三、启用 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 --zone=public --add-port=5672/tcp --permanent

firewall-cmd --zone=public --add-port=15672/tcp --permanent

firewall-cmd --zone=public --add-port=1883/tcp --permanent

firewall-cmd --zone=public --add-port=15675/tcp --permanent

如果搭建rabbitmq集群模式需要把下面这个两个端口放开

firewall-cmd --zone=public --add-port=4369/tcp --permanent

firewall-cmd --zone=public --add-port=25672/tcp --permanent

重启防火墙

systemctl restart firewalld

firewall-cmd --zone=public --list-ports

以上部署操作全部设定完毕，重启rabbitmq服务，使用创建root用户登录进入rabbitmq控制台

*至此rabbitmq搭建mqtt安装部署结束，下面进入代码实现环节

四、代码实现

先在pom中添加依赖包

<dependency>
     <groupId>org.springframework.integration</groupId>
     <artifactId>spring-integration-mqtt</artifactId>
 </dependency>
 <dependency>
     <groupId>org.eclipse.paho</groupId>
     <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
     <version>1.2.5</version>
 </dependency>

application.yml的所需的配置

mqtt:
  #MQTT-用户名 root
  username: third_client
  #MQTT-密码,需要解密 root
  password: OP74X53Z
  #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: 3000

mqtt服务类

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;

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 *mqtt网关（发布端需要用到）
 */
@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {


    /**
     * 发送到mqtt
     *
     * @param payload 有效载荷
     */
    void sendToMqtt(String payload);

    /**
     * 发送到mqtt
     *
     * @param topic   主题
     * @param payload 消息内容
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

    /**
     * 发送到mqtt
     *
     * @param topic   主题
     * @param qos     qos
     * @param payload 消息内容
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}

测试发送和订阅

五、mqtt.fx连接mqtt工具使用

mqtt测试工具安装包：https://pan.baidu.com/s/1oun7rMVJITOK9VSyO785HQ
提取码：l3cm

1、配置连接及订阅

配置mqtt连接

配置用户名密码

订阅gps-topic

六、jmeter压测结果展示

这里拟2w的消息并发量，根据业务计算最高模拟测试9w消息并发量没出现问题

如何使用移步到 ：https://blog.csdn.net/weixin_39393393/article/details/116640867?spm=1001.2014.3001.5502

1、使用jmeter模拟2w并发量

2、结果展示

rabbitmq控制台展示

后台打印

mqtt订阅的消息

jmeter压测报告