在Spring Boot中使用MQTT实现消息的发布和订阅，你可以使用Spring Integration with MQTT。以下是一个简单的例子：
- 添加依赖到你的
pom.xml：
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
</dependencies>- 配置MQTT客户端并定义消息通道：
@Configuration
public class MqttConfig {
@Value("${mqtt.broker.url}")
private String brokerUrl;
@Value("${mqtt.client.id}")
private String clientId;
@Value("${mqtt.username}")
private String userName;
@Value("${mqtt.password}")
private String password;
@Value("${mqtt.default.topic}")
private String defaultTopic;
@Bean
public MqttPahoClientFactory mqttClient() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{brokerUrl});
options.setUserName(userName);
options.setPassword(password.toCharArray());
factory.setConnectionOptions(options);
return factory;
}
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClient(), defaultTopic);
adapter.setCompletionTimeout(5000);
adapter.setQos(2);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
}- 消息处理：
@Component
@Log4j
public class MqttMessageHandler {
@Autowired
private SimpMessagingTemplate simpMessagingTemplate;
@Autowired
private MessageChannel mqttInputChannel;
@Autowired
private MqttPahoClientFactory mqttClientFactory;
@PostConstruct
public void init() {