A multiple-recipient message sends one event from a microservice to many independent consumers. It suits application-wide notifications, such as reporting that a video was viewed, while reducing coupling between services. section 5.8.7 section 5.8.9
Implement the pattern with a RabbitMQ exchange and one queue per recipient:
- Declare a fanout exchange, such as viewed. section 5.8.7
- Create an exclusive anonymous queue for each consuming microservice. section 5.8.7
- Bind each queue to the exchange and consume messages independently. section 5.8.7
- Publish the serialized event to the exchange instead of addressing a specific queue. section 5.8.7
Viewed-video notification
A video-streaming service publishes a viewed event. History stores the video path, while recommendations can independently receive the same event and use it for future suggestions. Run the working project with cd chapter-5/example-4 followed by docker compose up --build. section 5.8.7 section 5.8.9
Listing 5.14 Consuming “Viewed” messages from a RabbitMQ exchange
--snip-- const historyCollection = db.collection("history"); await messageChannel.assertExchange("viewed", "fanout"); #1 const { queue } = await messageChannel.assertQueue("", { exclusive: true }); #2 await messageChannel.bindQueue(queue, "viewed", ""); #3 await messageChannel.consume(queue, async (msg) { #4 const parsedMsg = JSON.parse(msg.content.toString()); #5 await historyCollection.insertOne({ #6 videoPath: parsedMsg.videoPath #6 }); #6 messageChannel.ack(msg); #7 }); --snip-- #1 Asserts the existence of the “Viewed” message exchange. The “fanout” option spreads the message out to the anonymous queues. #2 Creates an anonymous queue. The option exclusive is set to true so that the queue will be deallocated automatically when the microservice disconnects from it (otherwise, our application has a memory leak). Extracts the generated name of the anonymous queue. #3 Binds the queue to the exchange #4 Receives messages from the anonymous queue #5 Parses the JSON message to a JavaScript object #6 Records the view in the history database #7 If there is no error, acknowledges the message
Inspect the consumer configuration: it asserts a fanout exchange, creates an exclusive queue, binds it, parses the message, persists data, and acknowledges delivery. section 5.8.7
