{"id":25214,"date":"2026-06-01T06:38:56","date_gmt":"2026-05-31T23:38:56","guid":{"rendered":"https:\/\/huongdanjava.com\/?p=25214"},"modified":"2026-06-01T06:39:33","modified_gmt":"2026-05-31T23:39:33","slug":"consumer-routing-with-spring-cloud-stream","status":"publish","type":"post","link":"https:\/\/huongdanjava.com\/consumer-routing-with-spring-cloud-stream.html","title":{"rendered":"Consumer routing with Spring Cloud Stream"},"content":{"rendered":"

Spring Cloud Stream supports consumer routing, allowing us to decide which consumer will consume messages sent by a particular producer. We can route messages using message headers, payload content, or header content with expressions. Let’s explore this in detail in this tutorial!<\/p>\n

First, we’ll create a new Spring Boot project with the Cloud Stream dependency as follows:<\/p>\n

\"\"<\/p>\n

Similar to the previous tutorial<\/a>, we also add the Spring Cloud Starter Stream Rabbit dependency:<\/p>\n

<dependency>\r\n  <groupId>org.springframework.cloud<\/groupId>\r\n  <artifactId>spring-cloud-starter-stream-rabbit<\/artifactId>\r\n<\/dependency><\/pre>\n
and define the RabbitMQ information in the application.yml file as follow:<\/p>\n
spring:\r\n  rabbitmq:\r\n    host: localhost\r\n    port: 5672\r\n    username: guest\r\n    password: guest<\/pre>\n
To use consumer routing with Spring Cloud Stream<\/strong>, enable it by defining the following properties:<\/p>\n
spring:\r\n  cloud:\r\n    function:\r\n      routing:\r\n        enabled: true\r\n      definition: functionRouter<\/pre>\n
By enabling this routing, Spring will automatically create a Function interface bean named `functionRouter` and a logical binding name `functionRouter-in-0`. You can use this logical binding name to bind to the exchange with the queue in RabbitMQ, and when a message is received by this `functionRouter` bean, it will route the message to the consumer we want.<\/p>\n
I will define the binding for sending and receiving messages as follows:<\/p>\n
spring:\r\n  cloud:\r\n    stream:\r\n      bindings:\r\n        sendMessage:\r\n          destination: message.exchange\r\n        functionRouter-in-0:\r\n          destination: message.exchange\r\n          group: huongdanjava\r\n<\/pre>\n
As you can see, I’m using the StreamBridge class to send messages with the binding name sendMessage<\/em>, and I’ve also defined two consumers that can consume messages using the Consumer interface as follows:<\/p>\n
@Bean\r\npublic Consumer<String> receiveConsumer1() {\r\n  return message -> {\r\n    System.out.println(\"Received in consumer 1: \" + message);\r\n  };\r\n}\r\n\r\n@Bean\r\npublic Consumer<String> receiveConsumer2() {\r\n  return message -> {\r\n    System.out.println(\"Received in consumer 2: \" + message);\r\n  };\r\n}<\/pre>\n
Routing using message header<\/h3>\n
To route using message headers<\/strong>, use the MessageBuilder class to build the message as follows:<\/p>\n
Message<String> message = MessageBuilder\r\n    .withPayload(\"Hello World\")\r\n    .setHeader(\"spring.cloud.function.definition\", \"receiveConsumer1\")\r\n    .build();<\/pre>\n
The payload is the message we want to send. In the header, you must pass the header name spring.cloud.function.definition<\/em> with the header value being the name of the consumer bean that you want to receive the message. In this example, I’m configuring it as receiveConsumer1, as you can see!<\/p>\n
I also defined the bean of the CommandLineRunner interface to send messages using the StreamBridge class as follows:<\/p>\n
@Bean\r\nCommandLineRunner runner() {\r\n  return args -> {\r\n    Message<String> message = MessageBuilder\r\n        .withPayload(\"Hello World\")\r\n        .setHeader(\"spring.cloud.function.definition\", \"receiveConsumer1\")\r\n        .build();\r\n\r\n    streamBridge.send(\"sendMessage\", message);\r\n  };\r\n}<\/pre>\n
Now, if you run the application, you will see the following result:<\/p>\n
\"\"<\/p>\n
If you change the header value of the header name spring.cloud.function.definition<\/em> to receiveConsumer2<\/em>, you will see the message received from receiveConsumer2 as follows:<\/p>\n
\"\"<\/p>\n
Routing using expression<\/h3>\n
To route with expressions<\/strong>, you can use the `spring.cloud.function.routing-expression` property, for example:<\/p>\n
spring:\r\n  cloud:\r\n    function:\r\n      routing:\r\n        enabled: true\r\n      definition: functionRouter\r\n      routing-expression: \"new String(payload).contains('2') ? 'receiveConsumer2' : 'receiveConsumer1'\"\r\n    stream:\r\n      bindings:\r\n        sendMessage:\r\n          destination: message.exchange\r\n        functionRouter-in-0:\r\n          destination: message.exchange\r\n          group: huongdanjava\r\n<\/pre>\n
so, if you send a message containing the number 2:<\/p>\n
@Bean\r\nCommandLineRunner runner() {\r\n  return args -> {\r\n    Message<String> message = MessageBuilder\r\n        .withPayload(\"Hello 2)\r\n        .build();\r\n\r\n    streamBridge.send(\"sendMessage\", message);\r\n  };\r\n}<\/pre>\n
then you will see the following result:<\/p>\n
\"\"<\/p>\n
Change the message payload to “Hello 1” and you’ll see that consumer 1 receives the message!<\/p>\n
You can also define expressions with the message header<\/strong>, for example:<\/p>\n
spring:\r\n  cloud:\r\n    function:\r\n      routing-expression: \"headers['eventType']\"\r\n<\/pre>\n
khi \u0111\u00f3, c\u00e1c b\u1ea1n ch\u1ec9 c\u1ea7n truy\u1ec1n message header eventType v\u1edbi gi\u00e1 tr\u1ecb l\u00e0 t\u00ean c\u1ee7a consumer s\u1ebd nh\u1eadn message l\u00e0 \u0111\u01b0\u1ee3c:<\/p>\n
@Bean\r\nCommandLineRunner runner() {\r\n  return args -> {\r\n    Message<String> message = MessageBuilder\r\n        .withPayload(\"Hello World\")\r\n        .setHeader(\"eventType\", \"receiveConsumer1\")\r\n        .build();\r\n\r\n    streamBridge.send(\"sendMessage\", message);\r\n  };\r\n}<\/pre>\n
So, in this tutorial, I’ve shown you how to route messages to specific consumers. Use them appropriately depending on your needs!<\/p>\n
Watch the video here:<\/strong><\/p>\n