Spring Cloud Stream Kafka批次-手动提交整个批次

Question

我们正在使用Spring云流霍克斯顿。SR4使用来自Kafka主题的消息。我们启用了spring.cloud.stream.bindings.。consumer.batch-Mode=true，每次轮询获取2000条记录。我想知道是否有一种方法可以手动确认/提交整个批次。

Answer 1

有没有一种方法可以在批处理消费者中检索包含标题的消息列表，类似的（更新了上面的示例）

@Bean
Consumer<List<Message<Foo>>> consume() {
    return list -> {
        list.forEach(msg -> {
            System.out.println(msg.getPayload());
        });
    };
}

虽然我的问题是关于批处理消耗与头（如KafkaHeaders。MESSAGE_KEY）作为我的生产者发送所需的数据部分在关键和有效载荷Rest。我没有找到离我要找的最近的这条线。

Answer 2

SR4非常古老；当前霍克斯顿版本是SR9，当前Spring云流版本是3.0.10。释放（Hoxton.SR9拉入3.0.9）。

您需要使用Message并从标头获取确认。

@SpringBootApplication
public class So652289261Application {

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

    @Bean
    Consumer<Message<List<Foo>>> consume() {
        return msg -> {
            System.out.println(msg.getPayload());
            msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();
        };
    }

    @Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
        return (container, dest, group) -> container.getContainerProperties()
                .setCommitLogLevel(LogIfLevelEnabled.Level.INFO);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
        return args -> {
            template.send("consume-in-0", "{\"bar\":\"baz\"}".getBytes());
            template.send("consume-in-0", "{\"bar\":\"qux\"}".getBytes());
        };
    }

    public static class Foo {

        private String bar;

        public Foo() {
        }

        public Foo(String bar) {
            this.bar = bar;
        }

        public String getBar() {
            return this.bar;
        }

        public void setBar(String bar) {
            this.bar = bar;
        }

        @Override
        public String toString() {
            return "Foo [bar=" + this.bar + "]";
        }

    }

}

启动2.3.6和Cloud Hoxton的属性。SR9

spring.cloud.stream.bindings.consume-in-0.group=so65228926
spring.cloud.stream.bindings.consume-in-0.consumer.batch-mode=true
spring.cloud.stream.kafka.bindings.consume-in-0.consumer.auto-commit-offset=false

spring.kafka.producer.properties.linger.ms=50

Boot 2.4.0和Cloud 2020.0.0-M6的属性

spring.cloud.stream.bindings.consume-in-0.group=so65228926
spring.cloud.stream.bindings.consume-in-0.consumer.batch-mode=true
spring.cloud.stream.kafka.bindings.consume-in-0.consumer.ack-mode=MANUAL

spring.kafka.producer.properties.linger.ms=50

[Foo [bar=baz], Foo [bar=qux]]
... Committing: {consume-in-0-0=OffsetAndMetadata{offset=14, leaderEpoch=null, metadata=''}}