Spring Integration mqtt示例代码

下面是一个简单的Spring Integration MQTT示例代码：

@Configuration
@EnableIntegration
public class MqttConfiguration {

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

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

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

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

    @Value("${mqtt.client.id}")
    private String clientId;

    @Value("${mqtt.qos}")
    private int qos;

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setCleanSession(true);
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        return mqttConnectOptions;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

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

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                clientId,
                mqttClientFactory(),
                topic);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(qos);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println(message.getPayload());
            }
        };
    }

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

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                clientId + "_out",
                mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(qos);
        return messageHandler;
    }

    @Bean
    public IntegrationFlow mqttOutFlow() {
        return IntegrationFlows
                .from(mqttOutboundChannel())
                .handle(mqttOutbound())
                .get();
    }

}

其中，`mqtt.broker.url`、`mqtt.broker.username`、`mqtt.broker.password`、`mqtt.topic`、`mqtt.client.id`、`mqtt.qos`这些都是配置参数，需要在配置文件中进行配置。在这个示例代码中，我们使用了两个消息通道，`mqttInputChannel`和`mqttOutboundChannel`，用于接收和发送消息。使用`MqttPahoMessageDrivenChannelAdapter`来实现接收MQTT消息，使用`MqttPahoMessageHandler`来实现发送MQTT消息。在`handler()`方法中，我们将消息输出到控制台上。在`mqttOutbound()`方法中，我们配置发送消息的相关属性。

需要注意的是，这个示例代码并没有提供完整的配置，需要根据自己的实际情况进行修改和完善。