MQTT是什么

MQTT由IBM公司开发，是一个即时通讯协议，也是一个物联网传输协议，主要用于轻量级的订阅/发布式的消息传输。其设计目的主要是为低带宽和不稳定网络环境下的物联网设备提供服务。

MQTT中的概念

• 订阅(Subscribtion): 订阅包含主题筛选器（Topic Filter）和最大服务质量（QoS）。订阅会与一个会话（Session）关联。一个会话可以包含多个订阅。每一个会话中的每个订阅都有一个不同的主题筛选器
• 会话(Session): 每个客户端与服务器建立连接后就是一个会话，客户端和服务器之间有状态交互。会话存在于一个网络之间，也可能在客户端和服务器之间跨越多个连续的网络连接。
• 主题名(Topic Name): 连接到一个应用程序消息的标签，该标签与服务器的订阅相匹配。服务器会将消息发送给订阅所匹配标签的每个客户端。 需要注意的是，MQTT中消息主题按照层级命名，使用 ‘/’ 进行分割 此外，主题中可以使用通配符进行多个主题或多层级的订阅，有两种常见的通配符： 单层通配符 +：单层通配符只能匹配一层的主题，例如：China/Beijing/+，可以匹配的只有Beijing这个主题下面一层的主题，例如Xicheng, DongCheng, Xuanwu等等。 多层通配符 #：顾名思义，多层通配符就是可以匹配多个层级的主题，例如：China/#，可以匹配到的主题可能有：China/Beijing/Dongcheng, China/Shanghai/PuDong，等等。
• 主题筛选器(Topic Filter): 一个对主题名通配符筛选器，在订阅表达式中使用，表示订阅所匹配到的多个主题。
• 负载(Payload): 消息订阅者所具体接收的内容

MQTT协议的使用

MQTT是一个轻量的发布订阅模式消息传输协议，专门针对低带宽和不稳定网络环境的物联网应用设计。

特点

• 开放消息协议，简单易实现
• 发布订阅模式，一对多消息发布
• 基于TCP/IP网络连接
• 1字节固定报头，2字节心跳报文，报文结构紧凑
• 消息QoS支持，可靠传输保证

MQTT协议基于主题(Topic)进行消息路由，主题(Topic)类似URL路径

使用MQTT连接EMQ

生产端

@Configuration
public class MqttSenderConfig {

    @Value("${emq.connection.mqtt_url}")
    private String MQTT_URL;

    @Value("${emq.connection.client_token}")
    private String TOKEN;

    @Value("${emq.connection.client_id}")
    private String CLIENT_ID;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{MQTT_URL});
        options.setUserName(CLIENT_ID);
        options.setPassword(TOKEN.toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(CLIENT_ID, mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic("topic");
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }
}

@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MyGateway {

    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
    
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, @Header(MqttHeaders.RETAINED) Boolean retained, String payload);
}

retained true 保留数据，Broker会存储每个Topic的最后一条保留消息及其Qos，当订阅该Topic的客户端上线后，Broker需要将该消息投递给它。

消费端

@Configuration
@Slf4j
public class MqttInboundConfiguration {
    
    @Value("${emq.connection.mqtt_url}")
    private String MQTT_URL;

    @Value("${emq.connection.client_token}")
    private String TOKEN;

    @Value("${emq.connection.client_id}")
    private String CLIENT_ID;
    
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new   MqttPahoMessageDrivenChannelAdapter(MQTT_URL, CLIENT_ID,
                        "topic");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
    
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
}

@Slf4j
@Component
public class MqttSubscribeImpl {
    
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) throws MessagingException {
        log.info(message.getPayload().toString());
    }