spring-integration-mqtt 消息处理多线程

在使用 Spring Integration MQTT 消息处理时，可以通过配置适当的线程池来实现多线程处理。

首先，可以使用 Spring 的 TaskExecutor 接口定义一个线程池，例如：

@Bean
public TaskExecutor mqttTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100);
    return executor;
}

然后，在使用 MQTT Inbound Channel Adapter 时，可以设置其 executor 属性来指定使用上述线程池进行消息处理，例如：

<int-mqtt:inbound-channel-adapter id="mqttInbound"
    auto-startup="true"
    url="tcp://localhost:1883"
    client-id="mqttClient"
    topics="test_topic"
    qos="1"
    task-executor="mqttTaskExecutor"
    payload-expression="@mqttMessageConverter.fromMessage(#this)"
    channel="mqttChannel" />

这样，当有消息到达时，Spring Integration 将会使用指定的线程池进行消息处理，从而实现多线程处理。