A RabbitMQ exchange gives microservices an indirect messaging target. A sender publishes to the exchange, and RabbitMQ routes the message to queues instead of connecting sender and receiver directly. section 5.8.2
Exchanges are especially useful for one-to-many notifications. A single service can publish an event such as a viewed video, while several services receive and process their own copies. section 5.8.7
A fanout exchange workflow consists of these practical steps:
- Assert the shared exchange with the `fanout` type. section 5.8.7
- Create an anonymous exclusive queue for each receiving microservice. section 5.8.7
- Bind each queue to the exchange, then consume messages and acknowledge them after processing. section 5.8.7
- Publish serialized messages to the exchange with `messageChannel.publish`. section 5.8.7
Broadcasting a viewed-video event
The video-streaming service publishes `{ videoPath }` to the `viewed` exchange. The history and recommendations services each receive the event through their own queues and can handle it independently. section 5.8.7
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 that asserts the exchange, creates an exclusive queue, binds it, consumes messages, and acknowledges completed work. section 5.8.7
Listing 5.15 Publishing “Viewed” messages to a RabbitMQ exchange
--snip-- async function main() { const messagingConnection = await amqp.connect(RABBIT); const messageChannel = await messagingConnection.createChannel(); await messageChannel.assertExchange( #1 "viewed", #1 "fanout" #1 ); #1 function broadcastViewedMessage(messageChannel, videoPath) { #2 const msg = { videoPath: videoPath }; #3 const jsonMsg = JSON.stringify(msg); #4 messageChannel.publish( #5 "viewed", "", Buffer.from(jsonMsg) ); } app.get("/video", async (req, res) => { --snip-- sendViewedMessage(messageChannel, videoPath); #6 }); --snip-- } #1 Asserts the existence of the “Viewed” message exchange #2 A helper function to send the “Viewed” message #3 Defines the payload of the message #4 Converts the message to the JSON format #5 Publishes the message to the “Viewed” exchange #6 Sends the view message in response to the video being streamed to the web browser
Inspect the publisher configuration that asserts the exchange, serializes the event, and publishes it through amqplib. section 5.8.7
