SpringBoot集成Mqtt发送消息

news2024/9/22 22:34:47

1. MQTT简介

MQTT是一种物联网消息协议，为Message Queuing Telemetry Transport的缩写，即消息队列传输探测，协议基于发布订阅模式进行通信，有开销低、带宽小、轻量的特点，通常应用在物联网数据采集、移动应用、智能硬件、电力、能源等领域。

2. SpringBoot集成Mqtt

Spring集成Mqtt常用的有两种方式，一种是直接使用Mqtt的客户端库，如Eclipse Paho，另外一种是spring integration mqtt
第一种：使用Mqtt客户端库
依赖引入：org.eclipse.paho.client.mqttv3

<dependency>
	<groupId>org.eclipse.paho</groupId>
	<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
	<version>1.2.0</version>
</dependency>

服务端配置

public class MqttSendMsgService {
    private static String clientId = "test";
    private static String username = "admin";
    private static String password = "xxxxxx";
    private static String broker = "tcp://xxxxx:1883";
    public ReturnT<String> mqttSend(String param) {
        MqttClient client;
        try {
            client = new MqttClient(broker, clientId, new MemoryPersistence());
            client.setCallback(new MqttCallback() {
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost: " + cause.getMessage());
                }
                @Override
                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                    System.out.println("Message arrived: " + mqttMessage.getPayload());
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println("Delivery complete");
                }
            });

            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName(username);
            connOpts.setPassword(password.toCharArray());

            client.connect(connOpts);
            log.info("Connected to MQTT Broker!");


            //主题
            String topic="test/simple";
            //消息
            String content="发送测试";

            MqttMessage message = new MqttMessage();
            message.setQos(1);
            message.setRetained(false);
            message.setPayload(content.getBytes());
            //消息发送
            client.publish(topic,message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
        return ReturnT.SUCCESS;
    }
}

上面这种使用起来比较简单，生产环境使用最多的还是下面这种

第二种：使用 Spring integration进行集成，这里以发送消息为例
依赖引入

<dependency>
	<groupId>org.springframework.integration</groupId>
	<artifactId>spring-integration-mqtt</artifactId>
	<version>5.5.14</version>
</dependency>

添加yaml配置

mqtt.url = tcp://xxxxx:1883
mqtt.username = admin
mqtt.password = 123456
mqtt.clientId = test
mqtt.defaultTopic = /test/send
mqtt.keepAliveInterval = 60
mqtt.automaticReconnect = true
mqtt.cleanSession = false
mqtt.connectionTimeout = 30
mqtt.maxInflight = 1024

添加对应的属性配置类

@Component
public class MqttConfigProperties {
    @Value("${mqtt.url}")
    private String url;
    @Value("${mqtt.username}")
    private String username;
    @Value("${mqtt.password}")
    private String password;
    @Value("${mqtt.clientId}")
    private String clientId;
    @Value("${mqtt.defaultTopic}")
    private String defaultTopic;
    @Value("${mqtt.keepAliveInterval}")
    private Integer keepAliveInterval;
    @Value("${mqtt.automaticReconnect}")
    private Boolean automaticReconnect;
    @Value("${mqtt.cleanSession}")
    private Boolean cleanSession;
    @Value("${mqtt.connectionTimeout}")
    private Integer connectionTimeout;
    @Value("${mqtt.maxInflight}")
    private Integer maxInflight;
}

创建客户端配置类

@Configuration
@IntegrationComponentScan
public class MqttConfig {
    @Autowired
    private MqttConfigProperties mqttConfigProperties;

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        log.info("初始化mqtt信息{}", JSON.toJSON(mqttConfigProperties));
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqttConfigProperties.getUsername());
        options.setPassword(mqttConfigProperties.getPassword().toCharArray());
        options.setServerURIs(new String[]{mqttConfigProperties.getUrl()});
        options.setKeepAliveInterval(mqttConfigProperties.getKeepAliveInterval());
        options.setAutomaticReconnect(mqttConfigProperties.getAutomaticReconnect());
        options.setCleanSession(mqttConfigProperties.getCleanSession());
        options.setConnectionTimeout(mqttConfigProperties.getConnectionTimeout());
        options.setMaxInflight(mqttConfigProperties.getMaxInflight());
        return options;
    }
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory(MqttConnectOptions mqttConnectOptions) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions);
        return factory;
    }

    // 推送通道
    @Bean
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }
    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler sendHandler(MqttPahoClientFactory mqttPahoClientFactory) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfigProperties.getClientId() + "-publish", mqttPahoClientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(1);
        messageHandler.setDefaultTopic(mqttConfigProperties.getDefaultTopic());
        log.info("初始化mqttOutputChannel...");
        return messageHandler;
    }


}

发送网关接口

@MessagingGateway(defaultRequestChannel = "mqttOutputChannel")
public interface MqttGateway {
    /**
     * 发送消息
     *
     * @param topic
     * @param data
     */
    void send(@Header(MqttHeaders.TOPIC) String topic, String data);
}

这样，在发送消息时，直接将消息网关注入，调用发送方法就可以发送了

mqttGateway.send(topic, JSONObject.toJSONString(msg));

参考：
https://mqtt.org/

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1464036.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！