SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2510: RabbitMQ Super Streams Concurrency

Browse Source 
Add support for concurrency.

Use try with resources in stream tests.

Fix Test with Latest Boot

Resolves #2524
This commit is contained in:
Gary Russell
2022-10-05 13:22:16 -04:00
committed by Oleg Zhurakousky
parent c617f7e841
commit d93a4a0c4a
7 changed files with 198 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -315,9 +315,10 @@ public class RabbitExchangeQueueProvisioner
String routingKey = properties.getExtension().getBindingRoutingKey();
String rk = routingKey == null ? name : routingKey;
SuperStream ss = new SuperStream(name, properties.getInstanceCount(), (q, i) -> IntStream.range(0, i)
.mapToObj(j -> rk + "-" + j)
.collect(Collectors.toList()));
SuperStream ss = new SuperStream(name, properties.getInstanceCount() * properties.getConcurrency(),
(q, i) -> IntStream.range(0, i)
.mapToObj(j -> rk + "-" + j)
.collect(Collectors.toList()));
synchronized (this.autoDeclareContext) {
if (!this.autoDeclareContext.containsBean(name + ".superStream")) {
this.autoDeclareContext.getBeanFactory().registerSingleton(name + ".superStream", ss);

2
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -545,7 +545,7 @@ public class RabbitMessageChannelBinder extends
RabbitConsumerProperties extension) {
if (extension.getContainerType().equals(ContainerType.STREAM)) {
return StreamUtils.createContainer(consumerDestination, group, properties, destination, extension,
return StreamUtils.createContainer(consumerDestination, group, properties, destination,
getApplicationContext());
}
boolean directContainer = extension.getContainerType()

8
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java
View File

@@ -32,6 +32,7 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerP
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties.ProducerType;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.outbound.RabbitStreamMessageHandler;
@@ -72,7 +73,9 @@ public final class StreamUtils {
*/
public static MessageListenerContainer createContainer(ConsumerDestination consumerDestination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties, String destination,
RabbitConsumerProperties extension, AbstractApplicationContext applicationContext) {
ApplicationContext applicationContext) {
RabbitConsumerProperties extension = properties.getExtension();
StreamListenerContainer container = new StreamListenerContainer(applicationContext.getBean(Environment.class)) {
@@ -94,7 +97,8 @@ public final class StreamUtils {
container.setStreamConverter(applicationContext.getBean(beanName, StreamMessageConverter.class));
}
if (properties.getExtension().isSuperStream()) {
container.superStream(consumerDestination.getName(), consumerDestination.getName() + "." + group);
container.superStream(consumerDestination.getName(), consumerDestination.getName() + "." + group,
properties.getConcurrency());
}
return container;
}

2
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java
View File

@@ -29,7 +29,7 @@ public class RabbitTestContainer {
private static final RabbitMQContainer RABBITMQ;
static {
String image = "rabbitmq:management";
String image = "rabbitmq:3.11-management";
String cache = System.getenv().get("IMAGE_CACHE");
if (cache != null) {
image = cache + image;

118
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java
View File

@@ -20,7 +20,6 @@ import com.rabbitmq.stream.ConsumerBuilder;
import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.OffsetSpecification;
import com.rabbitmq.stream.ProducerBuilder;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;
@@ -29,7 +28,6 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.Binding;
@@ -66,78 +64,71 @@ public class RabbitStreamBinderModuleTests {
private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
private ConfigurableApplicationContext context;
@AfterEach
void tearDown() {
if (context != null) {
context.close();
context = null;
@Test
void testStreamContainer() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0")) {
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitConsumerProperties rProps = new RabbitConsumerProperties();
rProps.setContainerType(ContainerType.STREAM);
ExtendedConsumerProperties<RabbitConsumerProperties> props =
new ExtendedConsumerProperties<RabbitConsumerProperties>(rProps);
props.setAutoStartup(false);
Binding<MessageChannel> binding = rabbitBinder.bindConsumer("testStream", "grp", new QueueChannel(), props);
Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer");
assertThat(container).isInstanceOf(StreamListenerContainer.class);
((StreamListenerContainer) container).start();
verify(context.getBean(ConsumerBuilder.class)).offset(OffsetSpecification.first());
((StreamListenerContainer) container).stop();
}
}
@Test
void testStreamContainer() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0");
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitConsumerProperties rProps = new RabbitConsumerProperties();
rProps.setContainerType(ContainerType.STREAM);
ExtendedConsumerProperties<RabbitConsumerProperties> props =
new ExtendedConsumerProperties<RabbitConsumerProperties>(rProps);
props.setAutoStartup(false);
Binding<MessageChannel> binding = rabbitBinder.bindConsumer("testStream", "grp", new QueueChannel(), props);
Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer");
assertThat(container).isInstanceOf(StreamListenerContainer.class);
((StreamListenerContainer) container).start();
verify(this.context.getBean(ConsumerBuilder.class)).offset(OffsetSpecification.first());
((StreamListenerContainer) container).stop();
}
@Test
void testSuperStreamContainer() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0");
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitConsumerProperties rProps = new RabbitConsumerProperties();
rProps.setContainerType(ContainerType.STREAM);
rProps.setSuperStream(true);
ExtendedConsumerProperties<RabbitConsumerProperties> props =
new ExtendedConsumerProperties<RabbitConsumerProperties>(rProps);
props.setAutoStartup(false);
props.setInstanceCount(1);
Binding<MessageChannel> binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props);
Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer");
assertThat(container).isInstanceOf(StreamListenerContainer.class);
((StreamListenerContainer) container).start();
ConsumerBuilder builder = this.context.getBean(ConsumerBuilder.class);
verify(builder).singleActiveConsumer();
verify(builder).superStream("testSuperStream");
verify(builder).name("testSuperStream.grp");
((StreamListenerContainer) container).stop();
.run("--server.port=0")) {
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitConsumerProperties rProps = new RabbitConsumerProperties();
rProps.setContainerType(ContainerType.STREAM);
rProps.setSuperStream(true);
ExtendedConsumerProperties<RabbitConsumerProperties> props =
new ExtendedConsumerProperties<RabbitConsumerProperties>(rProps);
props.setAutoStartup(false);
props.setInstanceCount(1);
Binding<MessageChannel> binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props);
Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer");
assertThat(container).isInstanceOf(StreamListenerContainer.class);
((StreamListenerContainer) container).start();
ConsumerBuilder builder = context.getBean(ConsumerBuilder.class);
verify(builder).singleActiveConsumer();
verify(builder).superStream("testSuperStream");
verify(builder).name("testSuperStream.grp");
((StreamListenerContainer) container).stop();
}
}
@Test
void testStreamHandler() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0");
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitProducerProperties rProps = new RabbitProducerProperties();
rProps.setProducerType(ProducerType.STREAM_SYNC);
ExtendedProducerProperties<RabbitProducerProperties> props =
new ExtendedProducerProperties<RabbitProducerProperties>(rProps);
Binding<MessageChannel> binding = rabbitBinder.bindProducer("testStream", new DirectChannel(), props);
Object handler = TestUtils.getPropertyValue(binding, "val$producerMessageHandler");
assertThat(handler).isInstanceOf(RabbitStreamMessageHandler.class);
.run("--server.port=0")) {
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitProducerProperties rProps = new RabbitProducerProperties();
rProps.setProducerType(ProducerType.STREAM_SYNC);
ExtendedProducerProperties<RabbitProducerProperties> props =
new ExtendedProducerProperties<RabbitProducerProperties>(rProps);
Binding<MessageChannel> binding = rabbitBinder.bindProducer("testStream", new DirectChannel(), props);
Object handler = TestUtils.getPropertyValue(binding, "val$producerMessageHandler");
assertThat(handler).isInstanceOf(RabbitStreamMessageHandler.class);
}
}
@SpringBootApplication(proxyBeanMethods = false)
@@ -167,7 +158,6 @@ public class RabbitStreamBinderModuleTests {
}
@Bean
@ConditionalOnMissingBean
Environment rabbitStreamEnvironment(ConsumerBuilder consumerBuilder, ProducerBuilder producerBuilder) {
Environment env = mock(Environment.class);
given(env.consumerBuilder()).willReturn(consumerBuilder);

129
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream2/RabbitStreamBinderModuleIntegrationTests.java Normal file
View File

@@ -0,0 +1,129 @@
/*
* Copyright 2021-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.rabbit.stream2;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.rabbitmq.stream.Address;
import com.rabbitmq.stream.OffsetSpecification;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.amqp.EnvironmentBuilderCustomizer;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.rabbit.stream.listener.StreamListenerContainer;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Gary Russell
*/
public class RabbitStreamBinderModuleIntegrationTests {
private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
@Test
void testSuperStreamContainer() throws InterruptedException {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0")) {
BinderFactory binderFactory = context.getBean(BinderFactory.class);
RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null,
MessageChannel.class);
RabbitConsumerProperties rProps = new RabbitConsumerProperties();
rProps.setContainerType(ContainerType.STREAM);
rProps.setSuperStream(true);
ExtendedConsumerProperties<RabbitConsumerProperties> props =
new ExtendedConsumerProperties<RabbitConsumerProperties>(rProps);
props.setAutoStartup(false);
props.setInstanceCount(3);
props.setConcurrency(3);
Binding<MessageChannel> binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props);
Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer");
assertThat(container).isInstanceOf(StreamListenerContainer.class);
assertThat(container).extracting("concurrency").isEqualTo(3);
((StreamListenerContainer) container).start();
assertThat(context.getBean(SimpleProcessor.class).consumerCountLatch.await(10, TimeUnit.SECONDS)).isTrue();
RabbitAdmin admin = context.getBean(RabbitAdmin.class);
System.out.println(RABBITMQ.getMappedPort(15672));
for (int i = 0; i < 9; i++) {
Properties qProps = admin.getQueueProperties("testSuperStream-" + i);
assertThat(qProps).describedAs("Expected queue with index %d to exist", i).isNotNull();
}
((StreamListenerContainer) container).stop();
}
}
@SpringBootApplication(proxyBeanMethods = false)
public static class SimpleProcessor {
final CountDownLatch consumerCountLatch = new CountDownLatch(3);
@Bean
ConnectionFactory cf() {
return new CachingConnectionFactory(RABBITMQ.getMappedPort(5672));
}
@Bean
RabbitAdmin admin(ConnectionFactory cf) {
return new RabbitAdmin(cf);
}
@Bean
EnvironmentBuilderCustomizer environmenCustomizer() {
return env -> env
.addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552)));
}
@Bean
ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
return (cont, dest, group) -> {
if (cont instanceof StreamListenerContainer container) {
container.setConsumerCustomizer((id, builder) -> {
builder.consumerUpdateListener(context -> {
this.consumerCountLatch.countDown();
return OffsetSpecification.first();
});
});
}
};
}
}
}

5
docs/src/main/asciidoc/rabbit/rabbit_overview.adoc
View File

@@ -420,7 +420,7 @@ Default: `1`.
Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided.
To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`.
IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is also not supported at this time.
IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is supported for super streams only.
Only a single stream queue can be consumed by each binding.
To configure the binder to use `containerType=stream`, Spring Boot will automatically configure an `Environment` `@Bean` from the application properties.
@@ -493,12 +493,13 @@ 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
----
====
The framework will create a super stream named `super`, with 3 partitions.
The framework will create a super stream named `super`, with 9 partitions.
Up to 3 instances of this application can be deployed.
=== Advanced Listener Container Configuration