优化上面代码，使用异步线程池消费mqtt消息

可以在 `MqttConfig` 配置类中使用 `taskExecutor` 来创建一个异步线程池，然后将消息处理逻辑放入异步线程池中执行，从而实现异步消费 MQTT 消息。下面是优化后的代码：

@Configuration
public class MqttConfig {

    // 省略其他配置

    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttClientFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(brokerUrl, clientId + "_in", mqttClientFactory, "topic");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(mqttMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        adapter.setTaskExecutor(taskExecutor()); // 设置异步线程池
        return adapter;
    }

    @Bean
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(25);
        return executor;
    }

    @Bean
    public MessageHandler mqttMessageHandler() {
        return new MqttMessageHandler();
    }

    @Service
    public class MqttMessageHandler extends AbstractMessageHandler {
        @Override
        protected void handleMessageInternal(Message<?> message) throws Exception {
            String payload = (String) message.getPayload();
            taskExecutor().execute(() -> {
                // 处理 MQTT 消息
            });
        }
    }
}

上述代码中，`inbound` 方法中使用 `setTaskExecutor` 方法设置异步线程池；`taskExecutor` 方法创建一个 `ThreadPoolTaskExecutor` 实例，用于异步执行消息处理逻辑；`mqttMessageHandler` 中的 `handleMessageInternal` 方法中，将消息处理逻辑放入异步线程池中执行。

这样，就可以实现异步消费 MQTT 消息，提高应用程序的并发能力和响应速度。需要注意的是，线程池的配置需要根据实际情况进行调整，以充分利用系统资源并避免资源浪费。