Spring Boot整合MQTT EMQX的代码过程

好的，下面是Spring Boot整合MQTT EMQX的代码过程：

1. 首先，在pom.xml文件中添加下面的依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
        <version>5.2.5.RELEASE</version>
    </dependency>
</dependencies>

2. 在application.properties文件中添加EMQX的连接信息：

spring.mqtt.username=your_username
spring.mqtt.password=your_password
spring.mqtt.url=tcp://your_emqx_ip:1883

3. 创建一个MQTT消息的处理器，代码如下：

@Component
public class MqttMessageHandler {

    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

4. 创建一个MQTT连接的配置类，代码如下：

@Configuration
@EnableIntegration
public class MqttConfig {

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.url}")
    private String url;

    @Bean
    public DefaultMqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public IntegrationFlow mqttInFlow() {
        return IntegrationFlows.from(
                inboundAdapter(),
                c -> c.poller(Pollers.fixedDelay(1000))
        )
                .transform(p -> p)
                .handle("mqttMessageHandler", "handleMessage")
                .get();
    }

    private MessageProducerSupport inboundAdapter() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(url, "testClient", mqttClientFactory(), "testTopic");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        return adapter;
    }
}

5. 测试代码：

@RestController
public class TestController {

    @Autowired
    private MessageChannel mqttInputChannel;

    @GetMapping("/send")
    public void sendMqttMessage() {
        mqttInputChannel.send(MessageBuilder.withPayload("Hello MQTT").build());
    }
}

6. 运行Spring Boot应用程序，访问http://localhost:8080/send，可以在控制台上看到接收到的MQTT消息。

注意：在运行代码之前，请确保你已经安装并运行了EMQX，并且已经创建了一个名为“testTopic”的主题。