一、前言

本篇主要是围绕着MQTT这个点，在生产环境中可能存储某些主题随时订阅和取消的逻辑，因为自己本身项目的需要所以顺便把这部分补充一下

二、调整MQTT配置

1、调整 MqttConfig.java

在 MqttConfig.java 中新增两个方法

    /*** 添加订阅主题的方法** @param subscribeTopic 要订阅的主题*/public void addTopic(String subscribeTopic) {inbound().addTopic(subscribeTopic);System.out.println("Subscribed to new topic: " + subscribeTopic);}/*** 移除订阅主题的方法** @param unsubscribeTopic 要取消订阅的主题*/public void removeTopic(String unsubscribeTopic) {inbound().removeTopic(unsubscribeTopic);System.out.println("Unsubscribed from topic: " + unsubscribeTopic);}

2、新增 MqttDynamicSubscriberService.java

@Slf4j
@Service
public class MqttDynamicSubscriberService {/*** MQTT 配置类*/@Resourceprivate MqttConfig mqttConfig;/*** 动态增加订阅主题** @param topic 主题*/public void subscribeToNewTopic(String topic) {mqttConfig.addTopic(topic);}/*** 动态移除订阅主题** @param topic 主题*/public void unsubscribeFromTopic(String topic) {mqttConfig.removeTopic(topic);}
}

3、新增 MqttTopicDynamicController.java

@RestController
@RequestMapping("/mqtt/dynamic/topic")
public class MqttTopicDynamicController {/*** MQTT 动态订阅主题 Service 实现组件*/@Resourceprivate MqttDynamicSubscriberService mqttDynamicSubscriberService;/*** MQTT 动态增加订阅主题** @param topic 主题*/@PostMapping("/subscribe")public void subscribe(String topic) {mqttDynamicSubscriberService.subscribeToNewTopic(topic);}/*** MQTT 动态取消订阅主题** @param topic 主题*/@PostMapping("/unsubscribe")public void unsubscribe(String topic) {mqttDynamicSubscriberService.unsubscribeFromTopic(topic);}
}

4、补充多主题初始化问题

多主题消息订阅初始化，只要修改一下 MqttConfig#inbound 方法即可

    @Beanpublic MqttPahoMessageDrivenChannelAdapter inbound() {MqttPahoMessageDrivenChannelAdapter adapter =new MqttPahoMessageDrivenChannelAdapter(MqttClient.generateClientId(), mqttClientFactory(), "default-topic-one","default-topic-two","default-topic-three");  // 指定订阅的主题adapter.setCompletionTimeout(5000);adapter.setConverter(new DefaultPahoMessageConverter());adapter.setQos(0);  // 设置QoS级别adapter.setOutputChannel(mqttInputChannel());return adapter;}