MQTT协议#

1
@Data
2
@EqualsAndHashCode(callSuper = false)
3
@Component
4
@ConfigurationProperties(prefix = "rabbitmq.mqtt")
5
public class MqttConfig {
6
    /**
7
     * RabbitMQ连接用户名
8
     */
9
    private String username;
10
    /**
11
     * RabbitMQ连接密码
12
     */
13
    private String password;
14
    /**
15
     * RabbitMQ的MQTT默认topic
16
     */
17
    private String defaultTopic;
18
    /**
19
     * RabbitMQ的MQTT连接地址
20
     */
21
    private String url;
22
}

1
@Slf4j
2
@Configuration
3
public class MqttInboundConfig {
4
    @Autowired
5
    private MqttConfig mqttConfig;
6

7
    @Bean
8
    public MessageChannel mqttInputChannel() {
9
        return new DirectChannel();
10
    }
11

12
    @Bean
13
    public MessageProducer inbound() {
14
        MqttPahoMessageDrivenChannelAdapter adapter =
15
                new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getUrl(), "subscriberClient",
16
                        mqttConfig.getDefaultTopic());
17
        adapter.setCompletionTimeout(5000);
18
        adapter.setConverter(new DefaultPahoMessageConverter());
19
        //设置消息质量：0->至多一次；1->至少一次；2->只有一次
20
        adapter.setQos(1);
21
        adapter.setOutputChannel(mqttInputChannel());
22
        return adapter;
23
    }
24

25
    @Bean
26
    @ServiceActivator(inputChannel = "mqttInputChannel")
27
    public MessageHandler handler() {
28
        return new MessageHandler() {
29

30
            @Override
31
            public void handleMessage(Message<?> message) throws MessagingException {
32
                //处理订阅消息
33
                log.info("handleMessage : {}",message.getPayload());
34
            }
35

36
        };
37
    }
38
}

1
@Configuration
2
public class MqttOutboundConfig {
3

4
    @Autowired
5
    private MqttConfig mqttConfig;
6

7
    @Bean
8
    public MqttPahoClientFactory mqttClientFactory() {
9
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
10
        MqttConnectOptions options = new MqttConnectOptions();
11
        options.setServerURIs(new String[] { mqttConfig.getUrl()});
12
        options.setUserName(mqttConfig.getUsername());
13
        options.setPassword(mqttConfig.getPassword().toCharArray());
14
        factory.setConnectionOptions(options);
15
        return factory;
16
    }
17

18
    @Bean
19
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
20
    public MessageHandler mqttOutbound() {
21
        MqttPahoMessageHandler messageHandler =
22
                new MqttPahoMessageHandler("publisherClient", mqttClientFactory());
23
        messageHandler.setAsync(true);
24
        messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
25
        return messageHandler;
26
    }
27

28
    @Bean
29
    public MessageChannel mqttOutboundChannel() {
30
        return new DirectChannel();
31
    }
32
}

1
@Component
2
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
3
public interface MqttGateway {
4
    /**
5
     * 发送消息到默认topic
6
     */
7
    void sendToMqtt(String payload);
8

9
    /**
10
     * 发送消息到指定topic
11
     */
12
    void sendToMqtt(String payload, @Header(MqttHeaders.TOPIC) String topic);
13

14
    /**
15
     * 发送消息到指定topic并设置QOS
16
     */
17
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
18
}

1
@RestController
2
@RequestMapping("/mqtt")
3
public class MqttController {
4

5
    @Autowired
6
    private MqttGateway mqttGateway;
7

8
    /**
9
     * sendToDefaultTopic
10
     * @param payload
11
     */
12
    @PostMapping("/sendToDefaultTopic")
13
    public void sendToDefaultTopic(String payload) {
14
        mqttGateway.sendToMqtt(payload);
15
    }
16

17
    /**
18
     * sendToTopic
19
     * @param payload
20
     * @param topic
21
     */
22
    @PostMapping("/sendToTopic")
23
    public void sendToTopic(String payload, String topic) {
24
        mqttGateway.sendToMqtt(payload, topic);
25
    }
26
}