下面是一个快速的spring引导应用程序，展示了如何动态创建容器。

@SpringBootApplication
public class So61950229Application {

    public static void main(String[] args) {
        SpringApplication.run(So61950229Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(DynamicListener listener, KafkaTemplate<String, String> template) {
        return args -> {
            IntStream.range(0, 10).forEach(i -> template.send("so61950229", "foo" + i));
            System.out.println("Hit enter to start a listener");
            System.in.read();
            listener.newContainer("so61950229", 0);
            System.out.println("Hit enter to start another listener");
            System.in.read();
            listener.newContainer("so61950229", 0);
        };
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so61950229").partitions(1).replicas(1).build();
    }

}

@Component
class DynamicListener {

    private static final Logger LOG = LoggerFactory.getLogger(DynamicListener.class);

    private final ConcurrentKafkaListenerContainerFactory<String, String> factory;

    private final ConcurrentMap<String, AbstractMessageListenerContainer<String, String>> containers
            = new ConcurrentHashMap<>();

    DynamicListener(ConcurrentKafkaListenerContainerFactory<String, String> factory) {
        this.factory = factory;
    }

    void newContainer(String topic, int partition) {
        ConcurrentMessageListenerContainer<String, String> container =
                this.factory.createContainer(new TopicPartitionOffset(topic, partition));
        String groupId = UUID.randomUUID().toString();
        container.getContainerProperties().setGroupId(groupId);
        container.setupMessageListener((MessageListener) record -> {
            System.out.println(record);
        });
        this.containers.put(groupId, container);
        container.start();
    }

    @EventListener
    public void idle(ListenerContainerIdleEvent event) {
        AbstractMessageListenerContainer<String, String> container = this.containers.remove(
                event.getContainer(ConcurrentMessageListenerContainer.class).getContainerProperties().getGroupId());
        if (container != null) {
            LOG.info("Stopping idle container");
            container.stop(() -> LOG.info("Stopped"));
        }
    }

}

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.idle-event-interval=5000