|
对于最新的稳定版本，请使用 spring-cloud-stream 4.3.0！ |
RabbitMQ Stream插件的初始消费者支持
现在提供了对 RabbitMQ Stream 插件的基本支持。要启用此功能，您必须将spring-rabbit-streamjar 到类路径 - 它必须与spring-amqp和spring-rabbit.
当您将containerType属性设置为stream; concurrency仅支持超级流。每个绑定只能使用单个流队列。 |
配置要使用的活页夹containerType=stream，Spring Boot 将自动配置一个Environment @Bean从应用程序属性。可以选择添加定制器来自定义侦听器容器。
@Bean
ListenerContainerCustomizer<MessageListenerContainer> customizer() {
return (cont, dest, group) -> {
StreamListenerContainer container = (StreamListenerContainer) cont;
container.setConsumerCustomizer((name, builder) -> {
builder.offset(OffsetSpecification.first());
});
// ...
};
}这name传递给定制器的参数是destination + '.' + group + '.container'.
流name()（用于偏移跟踪）设置为绑定destination + '.' + group. 可以使用ConsumerCustomizer如上所示。如果您决定使用手动偏移跟踪，则Context可用作邮件头：
int count;
@Bean
public Consumer<Message<?>> input() {
return msg -> {
System.out.println(msg);
if (++count % 1000 == 0) {
Context context = msg.getHeaders().get("rabbitmq_streamContext", Context.class);
context.consumer().store(context.offset());
}
};
}有关配置环境和消费者构建器的信息，请参阅 RabbitMQ Stream Java 客户端文档。
对 RabbitMQ 超级流的消费者支持
有关超级流的信息，请参阅超级流。
使用超级流允许在超级流的每个分区上使用单个活动使用者进行自动纵向扩展、纵向缩减。
配置示例：
@Bean
public Consumer<Thing> input() {
...
}spring.cloud.stream.bindings.input-in-0.destination=super
spring.cloud.stream.bindings.input-in-0.group=test
spring.cloud.stream.bindings.input-in-0.consumer.instance-count=3
spring.cloud.stream.bindings.input-in-0.consumer.concurrency=3
spring.cloud.stream.rabbit.bindings.input-in-0.consumer.container-type=STREAM
spring.cloud.stream.rabbit.bindings.input-in-0.consumer.super-stream=true该框架将创建一个名为super，有 9 个分区。最多可以部署此应用程序的 3 个实例。