如何处理轮询消费者（PollableMessageSource）中的批处理记录？

Question

我正在将 PolledConsumer 与 spring 云流一起使用。 我的消费者看起来像这样：

@Bean
    public ApplicationRunner runner(PollableMessageSource input, MessageChannel output) {
        return args -> {
            System.out.println("Send some messages to topic polledConsumerIn and receive from polledConsumerOut");
            System.out.println("Messages will be processed one per second");
            exec.execute(() -> {
                boolean result = false;
                while (true) {
                    // this is where we poll for a message, process it, and send a new one
                    result = input.poll(m -> {
                        String payload = (String) m.getPayload();
                        System.out.println("Received: " + payload);
                        output.send(MessageBuilder.withPayload(payload.toUpperCase())
                            .copyHeaders(m.getHeaders())
                            .build());
                    }, new ParameterizedTypeReference<String>() { });

                    try {
                        Thread.sleep(1_000);
                    }
                    catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                    if (result) {
                        System.out.println("Success");
                    }
                }
            });
        };
    }

我正在尝试以批处理模式使用记录，我的目标是在轮询后获取记录列表，因为我的方法是批处理。 在代码中 input.poll 方法采用 MessageHandler，它接受一条记录作为参数。 在我这样设置配置后： 绑定： 人： 消费者： 批处理模式：真

   binder:
       configuration:
         max.poll.records: 1500
         fetch.min.bytes: 900000
         fetch.max.wait.ms: 5000
         value.deserializer: tr.PersonDeserializer

结果还是一样。

有什么方法可以处理 MessageHandler 中的记录列表，这意味着 m.getPayload 的类型是 List ？

Answer 1

目前轮询消费者不支持批量消费。