Trong bài viết trước, mình đã giới thiệu với các bạn về Request/Reply pattern và cũng đã làm một ví dụ implement pattern này với Correlation ID. Ở bài viết này, mình sẽ làm một ví dụ khác, cũng implement Request/Reply pattern nhưng sẽ sử dụng temporary queue thay vì Correlation ID các bạn nhé!
Đầu tiên, mình cũng sẽ tạo mới một Maven project để làm ví dụ:
Mình sẽ sử dụng RabbitMQ để làm Message Broker thay vì ActiveMQ Artemis cho ví dụ này.
RabbitMQ Client dependency được khai báo như sau:
|
1 2 3 4 5 |
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.25.0</version> </dependency> |
Với approach sử dụng temporary queue, chúng ta không cần tạo reply queue manually. Reply queue sẽ được tạo tự động, chỉ được sử dụng được 1 lần và nó sẽ tự động bị xoá sau khi sử dụng xong.
Bây giờ chúng ta sẽ implement chi tiết cho ví dụ này các bạn nhé!
Requester
Tương tự như khi làm việc với ActiveMQ Artemis, với RabbitMQ, điều đầu tiên chúng ta cần phải làm là lấy connection tới RabbitMQ server với factory pattern sử dụng class ConnectionFactory, như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
package com.huongdanjava.eda.requestreply; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Requester { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection()) { Channel channel = connection.createChannel(); } } } |
Khi làm việc với RabbitMQ, chúng ta sẽ cần phải làm việc với channel sau khi đã có connection tới RabbitMQ server. Nói thêm về channel thì nó giúp chúng ta không cần mở nhiều connection tới RabbitMQ server mà vẫn làm việc được với RabbitMQ server đó các bạn. Một connection, các bạn có thể tạo nhiều channel khác nhau!
Để tạo một temporary reply queue trong RabbitMQ server, các bạn có thể sử dụng phương thức queueDeclare() của đối tượng Channel như sau:
|
1 |
String replyQueue = channel.queueDeclare("", false, true, true, null).getQueue(); |
- Tham số thứ nhất trong phương thức queueDelare() là queue name. Nếu các bạn để empty như mình, tên queue name sẽ được tạo random.
- Tham số thứ hai trong phương thức queueDeclare() này là durable, giá trị là true nếu các bạn muốn queue này vẫn tồn tại sau khi RabbitMQ server restart. Ngược lại, nếu giá trị là false thì queue sẽ bị xoá.
- Tham số thứ ba là exclusive, giá trị true nghĩa là queue này chỉ available cho connection này, cho channel này.
Tham số thứ tư là autoDelete, giá trị true nghĩa là server sẽ tự động xoá queue nếu nó không được sử dụng. - Tham số thứ năm là về một số configuration khác cho queue.
Sau khi đã tạo temporary reply queue, bây giờ chúng ta sẽ gửi message sử dụng phương thức basicPublish() của đối tượng Channel như sau:
|
1 2 3 4 5 6 7 8 9 10 |
String correlationId = UUID.randomUUID().toString(); AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .correlationId(correlationId) .replyTo(replyQueue) .build(); String message = "Huong Dan Java"; channel.basicPublish("", "eda.request.queue", props, message.getBytes()); System.out.println("Sent request with correlation ID: " + correlationId); |
Để nhận message, chúng ta sẽ sử dụng phương thức basicConsume() của đối tượng Channel. Một trong những tham số trong phương thức này là interface DeliverCallback. Các bạn cần implement interface này để có thể nhận message được gửi tới reply queue của RabbitMQ server.
Mình sẽ implement interface này đơn giản như sau:
|
1 2 3 4 5 6 7 |
DeliverCallback deliverCallback = (consumerTag, delivery) -> { if (delivery.getProperties().getCorrelationId().equals(correlationId)) { String response = new String(delivery.getBody(), "UTF-8"); System.out.println("Received reply: " + response); } }; |
Phương thức basicConsume() sẽ được gọi như sau:
|
1 |
channel.basicConsume(replyQueue, true, deliverCallback, consumerTag -> {}); |
Toàn bộ code của Requester như sau các bạn nhé:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 |
package com.huongdanjava.eda.requestreply; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.UUID; import java.util.concurrent.TimeoutException; public class Requester { public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection()) { Channel channel = connection.createChannel(); String replyQueue = channel.queueDeclare("", false, true, true, null).getQueue(); String correlationId = UUID.randomUUID().toString(); AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .correlationId(correlationId) .replyTo(replyQueue) .build(); String message = "Huong Dan Java"; channel.basicPublish("", "eda.request.queue", props, message.getBytes()); System.out.println("Sent request with correlation ID: " + correlationId); DeliverCallback deliverCallback = (consumerTag, delivery) -> { if (delivery.getProperties().getCorrelationId().equals(correlationId)) { String response = new String(delivery.getBody(), "UTF-8"); System.out.println("Received reply: " + response); } }; channel.basicConsume(replyQueue, true, deliverCallback, consumerTag -> { }); Thread.sleep(5000); } } } |
Receiver
Phía receiver, chúng ta cũng implement tương tự như requester, chỉ có thêm một bước là sau khi nhận message từ requester, receiver sẽ gửi lại message response tới reply queue.
Toàn bộ code của Receiver, mình viết như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
package com.huongdanjava.eda.requestreply; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver { public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection()) { Channel channel = connection.createChannel(); channel.queueDeclare("eda.request.queue", false, false, false, null); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String request = new String(delivery.getBody(), "UTF-8"); System.out.println("Server received request: " + request); String response = String.format("Hello, %s!", request); AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .correlationId(delivery.getProperties().getCorrelationId()) .build(); System.out.println("Reply Queue: " + delivery.getProperties().getReplyTo()); channel.basicPublish("", delivery.getProperties().getReplyTo(), props, response.getBytes()); }; channel.basicConsume("eda.request.queue", true, deliverCallback, consumerTag -> { }); // Keep the server alive System.out.println("Server is listening..."); Thread.sleep(60000); } } } |
Như các bạn thấy, chúng ta cần phải lấy Correlation ID và reply queue từ request message.
Bây giờ nếu các bạn chạy Receiver trước:
sau đó thì chạy Requester, các bạn sẽ thấy kết quả như sau:
Kiểm tra console log của Receiver, các bạn sẽ thấy tên của reply queue như sau:
