How to handle each separate thread by MQTT topic name in Spring Integration?

Using Spring Integration, I created an application that processes received messages by subscribing to a specific topic of MQTT that delivers the status of a sensor.

Spring Integration Config:

@Configuration
@EnableIntegration
@IntegrationComponentScan
public class MqttConfig {

    @Autowired
    private MqttConnectionEnv mqttConnectionEnv;

    @Value("${mqtt.topic.root}")
    private String mqttRootTopic;
    private final int mqttQos = 2;

    private final String clientId = UUID.randomUUID().toString();

    private MqttConnectOptions connectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setServerURIs(new String[] { mqttConnectionEnv.getBrokerUri() });
        options.setUserName(mqttConnectionEnv.getUsername());
        options.setPassword(mqttConnectionEnv.getPassword().toCharArray());
        return options;
    }

    private MemoryPersistence memoryPersistence() {
        return new MemoryPersistence();
    }

    /**
     * MQTT client factory
     */
    @Bean
    public DefaultMqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(connectOptions());
        factory.setPersistence(memoryPersistence());
        return factory;
    }

    /*** input channel settings... ***/

    /**
     * input channel
     */
    @Bean
    public MessageChannel inputChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT channel adapter
     */
    @Bean
    public MessageProducerSupport mqttInboundChannel() {
        MqttPahoMessageDrivenChannelAdapter channelAdapter =
                new MqttPahoMessageDrivenChannelAdapter(this.clientId, mqttClientFactory(), this.mqttRootTopic);
        channelAdapter.setCompletionTimeout(5000);
        channelAdapter.setConverter(new DefaultPahoMessageConverter());
        channelAdapter.setQos(this.mqttQos);
        channelAdapter.setOutputChannel(inputChannel());
        channelAdapter.setRecoveryInterval(1000);
        return channelAdapter;
    }

    /**
     * MQTT input channel handler
     */
    @ServiceActivator
    public MessageHandler inboundMessageHandler() {
        return message -> {
            String topic = (String)message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
            String msg = (String)message.getPayload();
            String now = LocalDateTime.now().toString();

            String result = String.format("[%s] %s >> %s", now, topic, msg);

            System.out.println(result);
        };
    }

    /**
     * MQTT input channel flow
     */
    @Bean
    public IntegrationFlow mqttInboundFlow() {
        return IntegrationFlows
            .from(mqttInboundChannel())
            .handle(inboundMessageHandler())
            .get();
    }
}

In this case, it usually works well, but if the number of sensors increases significantly or the processing of inboundMessageHandler takes a long time, a bottleneck occurs and processing is delayed.

So, I tried using a separate thread in the event handler, but if the code of inboundMessageHandler is complicated, the order of data reception cannot be guaranteed.

/**
 * MQTT input channel handler
 */
@ServiceActivator
public MessageHandler inboundMessageHandler() {
    return message -> {
        String topic = (String)message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
        String msg = (String)message.getPayload();
        String now = LocalDateTime.now().toString();

        String result = String.format("[%s] %s >> %s", now, topic, msg);

        new Thead(() -> {
            // some process...
        }).start();
    };
}

So the idea I came up with is to assign an index that identifies the sensor to the topic name of MQTT, and use each thread for each topic name.

So the application subscribes to the /awesomeApp/# topic, and the handler when receiving the /awesomeApp/sensor/1 topic and the message handler when receiving the /awesomeApp/sensor/2 topic want to be handled in separate threads.

However, a new sensor can be added while the application is running, and when the reception of the /awesomeApp/sensor/3 topic corresponding to the new sensor is confirmed, I want to process the handler using a new thread corresponding to the topic.

If possible, can it be solved only with elements of Spring Integration without developing message queues within inboundMessageHandler?