From 63b52dda7696c36d379b04afb04c4cadc4eb0e7f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 31 Oct 2017 11:30:00 -0400 Subject: [PATCH] GH-101: Documentation for partitioning Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/101 --- .../src/main/asciidoc/building.adoc | 5 +- .../src/main/asciidoc/index.adoc | 1 + .../src/main/asciidoc/overview.adoc | 2 +- .../src/main/asciidoc/partitions.adoc | 107 ++++++++++++++++++ 4 files changed, 112 insertions(+), 3 deletions(-) create mode 100644 spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/building.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/building.adoc index 9ad502876..348b12555 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/building.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/building.adoc @@ -8,8 +8,9 @@ To build the source you will need to install JDK {jdkversion}. The build uses the Maven wrapper so you don't have to install a specific -version of Maven. To enable the tests, you should have Kafka server 0.9 or above running -before building. See below for more information on running the servers. +version of Maven. To enable the tests, you should have RabbitMQ server running +on localhost and the default port (5672) +before building. The main build command is diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc index 3d6cf97ea..a38c6f858 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc @@ -27,6 +27,7 @@ include::overview.adoc[] include::dlq.adoc[] +include::partitions.adoc[] = Appendices [appendix] diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index 701da4a54..a503493f4 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -462,7 +462,7 @@ Default: `no limit` [NOTE] ==== In the case of RabbitMQ, content type headers can be set by external applications. -Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport (including transports, such as Kafka, that do not normally support headers). +Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport (including transports, such as Kafka (prior to 0.11), that do not natively support headers). ==== == Retry With the RabbitMQ Binder diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc new file mode 100644 index 000000000..ba7511057 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc @@ -0,0 +1,107 @@ +== Partitioning with the RabbitMQ Binder + +RabbitMQ does support partitioning natively. + +Sometimes it is advantageous to send data to specific partitions, for example when you want to strictly order message processing - all messages for a particular customer should go to the same partition. + +The `RabbitMessageChannelBinder` provides partitioning by binding a queue for each partition to the destination exchange. + +The following illustrates how to configure the producer and consumer side: + +[source, java] +---- +@SpringBootApplication +@EnableBinding(Source.class) +public class RabbitPartitionProducerApplication { + + private static final Random RANDOM = new Random(System.currentTimeMillis()); + + private static final String[] data = new String[] { + "foo1", "bar1", "qux1", + "foo2", "bar2", "qux2", + "foo3", "bar3", "qux3", + "foo4", "bar4", "qux4", + }; + + public static void main(String[] args) { + new SpringApplicationBuilder(RabbitPartitionProducerApplication.class) + .web(false) + .run(args); + } + + @InboundChannelAdapter(channel = Source.OUTPUT, poller = @Poller(fixedRate = "5000")) + public Message generate() { + String value = data[RANDOM.nextInt(data.length)]; + System.out.println("Sending: " + value); + return MessageBuilder.withPayload(value) + .setHeader("partitionKey", value) + .build(); + } + +} +---- + +.application.yml +[source, yaml] +---- + spring: + cloud: + stream: + bindings: + output: + destination: partitioned.destination + producer: + partitioned: true + partition-key-expression: headers['partitionKey'] + partition-count: 2 + required-groups: + - myGroup +---- + +[NOTE] +==== +The above configuration uses the default partitioning (`key.hashCode() % partitionCount`). +This may or may not provide a suitably balanced algorithm, depending on the key values; you can override this default by using the `partitionSelectorExpression` or `partitionSelectorClass` properties. + +The `required-groups` property is only required if you need the consumer queues to be provisioned when the producer is deployed. +Otherwise, any messages sent to a partition will be lost until the corresponding consumer is deployed. +==== + +IMPORTANT: The `RabbitMessageChannelBinder` does not support dynamic scaling; there must be at least one consumer per partition. +The consumer's `instanceIndex` is used to indicate which partition will be consumed. +On platforms such as Cloud Foundry there can only be one instance with an `instanceIndex`. + +[source, java] +---- +@SpringBootApplication +@EnableBinding(Sink.class) +public class RabbitPartitionConsumerApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder(RabbitPartitionConsumerApplication.class) + .web(false) + .run(args); + } + + @StreamListener(Sink.INPUT) + public void listen(@Payload String in, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) { + System.out.println(in + " received from queue " + queue); + } + +} +---- + +.application.yml +[source, yaml] +---- + spring: + cloud: + stream: + bindings: + input: + destination: partitioned.destination + group: myGroup + consumer: + partitioned: true + instance-index: 0 +----