Sử dụng binding Exchange to Exchange trong RabbitMQ

1015

Bài viết được sự cho phép của tác giả Giang Phan

Trong các bài viết trước chúng ta đã cùng tìm hiểu về các loại Exchange trong RabbitMQ. Có một câu hỏi đặt ra là có thể thực hiện binding một Exchange đến Exchange khác hay không? Câu trả lời là có và chúng ta sẽ thấy cách thực hiện như thế nào trong phần tiếp theo của bài viết này.

  Kết nối RabbitMQ sử dụng Web STOMP Plugin
  Sử dụng publisher confirm trong RabbitMQ

Flow của một Message trong Exchange-to-Exchange bindings

Để dễ hiểu, tôi sẽ mô tả Flow của một Message được chuyển từ Topic Exchange sang Header Exchange.

Sơ đồ bên dưới là sự kết hợp của sơ đồ ở bài viết Topic Exchange và Header Exchange. Điểm khác biệt duy nhất là cách mà Header Exchange binding tới Topic Exchange với routing key pattern “#.gpcoder.com“.

Sử dụng binding Exchange to Exchange trong RabbitMQ

  • Một Producer publish một Message đến source Exchange với một routing key dựa trên loại của Exchange. Trong trường hợp này là GPCoderTopicExchange.
  • Có 2 Queue: QJava và QAll được binding tới Exchange GPCoderTopicExchange lần lượt với routing key patter:
    • QJava : Queue này sẽ nhận tất cả message có Key match với routing key “java.*.gpcoder.com“. Nghĩa là chỉ nhận các message cho một topic Java cụ thể từ gpcoder.com, chẳng hạn: java.core.gpcoder.com, java.collection.gpcoder.com.
    • QAll : Queue này nhận tất cả message có Key match với routing key “#.gpcoder.com“. Nghĩa là nhận tất cả message từ gpcoder.com, chẳng hạn: design-pattern.gpcoder, java.gpcoder.com, creational.design-pattern.gpcoder.com.
  • Một Exchange khác là GPCoderHeadersExchange binding tới Exchange GPCoderTopicExchange với routing key pattern #.gpcoder.com.
  • Có 3 Queue: QDeveloper, QManager và QPublished được binding tới Exchange GPCoderHeadersExchange với các header:
    • QDeveloper : Queue này sẽ nhận tất cả message có header là {“dev”, “Developer Channel”} hoặc {“general”, “General Channel”}.
    • QManager : Queue này nhận tất cả message có header là {“dev”, “Developer Channel”} hoặc {“general”, “General Channel”} hoặc {“manager”, “Manager Channel”}.
    • QPublished : Queue này nhận tất cả message có header là {“dev”, “Developer Channel”} và {“access”, “publish”}.
  • Bây giờ nếu một Message được gởi tới Exchange với routing key là design-pattern.gpcoder.com và header là {“dev”, “Developer Channel”} sẽ được chuyển tới Queue QAll và GPCoderHeadersExchange. Sau đó GPCoderHeadersExchange sẽ chuyển đến Queue QDeveloper và QManager dựa vào header {“dev”, “Developer Channel”}.
  • Tương tự vậy, chúng ta cũng có thể định nghĩa một hay nhiều Exchange khác, chẳng hạn GPCoderDirectExchange binding tới GPCoderHeadersExchange và GPCoderTopicExchange.

Một Exchange nhận Message từ publisher được gọi là Source Exchange. Trong sơ đồ trên, GPCoderTopicExchange là Source Exchange.

Một Exchange nhận Message từ một Exchange khác gọi là Destination Exchange. Trong sơ đồ trên, GPCoderHeadersExchange là Destination Exchange.

Ví dụ binding Exchange to Exchange trong RabbitMQ

Một số class của chương trình:

  • ConnectionManager : hỗ trợ tạo Connection đến RabbitMQ.
  • ExchangeChannel :  class util hỗ trợ tạo Echange, Queue, binding Queue đến Exchange, binding Exchange đến Exchange, publish/ subscribe message, …
  • Constant : định nghĩa constant chứa các thông tin về tên Exchange, Queue.
  • HeadersExchangeProducer : để gửi Message đến GPCoderHeadersExchange.
  • TopicExchangeProducer : để gửi Message đến GPCoderTopicExchange.
  • HeadersExchangeConsumer : để nhận Message từ Queue được binding đến GPCoderHeadersExchange.
  • TopicExchangeConsumer : để nhận Message từ Queue được binding đến GPCoderTopicExchange.
  • App: giả lập việc gửi nhận Message thông qua Topic Exchange của RabbitMQ.

ConnectionManager.java

package com.gpcoder.exchange2exchange;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConnectionManager {

private ConnectionManager() {
super();
}

public static Connection createConnection() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
return factory.newConnection();
}
}

ExchangeChannel.java

package com.gpcoder.exchange2exchange;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class ExchangeChannel {

private String exchangeName;
private Channel channel;
private Connection connection;

public ExchangeChannel(Connection connection, String exchangeName) throws IOException {
this.exchangeName = exchangeName;
this.connection = connection;
this.channel = connection.createChannel();
}

public void declareExchange(BuiltinExchangeType exchangeType) throws IOException {
// exchangeDeclare( exchange, builtinExchangeType, durable)
channel.exchangeDeclare(exchangeName, exchangeType, true);
}

public void declareQueues(String ...queueNames) throws IOException {
for (String queueName : queueNames) {
// queueDeclare - (queueName, durable, exclusive, autoDelete, arguments)
channel.queueDeclare(queueName, true, false, false, null);
}
}

public void performQueueBinding(String queueName, String routingKey, Map<String, Object> headers) throws IOException {
// Create bindings - (queue, exchange, routingKey, headers)
channel.queueBind(queueName, exchangeName, routingKey, headers);
}

public void performExchangeBinding(String destination, String source, String routingKey) throws IOException {
// (destination-exchange, source-exchange , routingKey
channel.exchangeBind(destination, source, routingKey);
}

public void subscribeMessage(String queueName) throws IOException {
// basicConsume - ( queue, autoAck, deliverCallback, cancelCallback)
channel.basicConsume(queueName, true, ((consumerTag, message) -> {
System.out.println("[Received] [" + queueName + "]: " + consumerTag);
System.out.println("[Received] [" + queueName + "]: " + new String(message.getBody()));
}), consumerTag -> {
System.out.println(consumerTag);
});
}

public void publishMessage(String message, String routingKey, Map<String, Object> headers) throws IOException {
BasicProperties properties = new BasicProperties()
.builder().headers(headers).build();

// basicPublish - ( exchange, routingKey, basicProperties, body)
System.out.println("[Send] [" + headers + "]: " + message);
channel.basicPublish(exchangeName, routingKey, properties, message.getBytes());
}
}

Constant.java

package com.gpcoder.exchange2exchange;

public final class Constant {

// Exchange

public static final String HEADERS_EXCHANGE_NAME = "GPCoderHeadersExchange";

public static final String TOPIC_EXCHANGE_NAME = "GPCoderTopicExchange";

// Queue

public static final String DEV_QUEUE_NAME = "QDeveloper";

public static final String MANAGER_QUEUE_NAME = "QManager";

public static final String PUBLISHED_QUEUE_NAME = "QPublished";

public static final String JAVA_QUEUE_NAME = "QJava";

public static final String ALL_QUEUE_NAME = "QAll";

// Routing key pattern

public static final String JAVA_ROUTING_KEY = "java.*.gpcoder.com";

public static final String GPCODER_ROUTING_KEY = "#.gpcoder.com";

// Message key

public static final String JAVA_CORE_MSG_KEY = "java.core.gpcoder.com";

public static final String JAVA_MSG_KEY = "java.gpcoder.com";

public static final String DESIGN_PATTERN_MSG_KEY = "design-pattern.gpcoder.com";

public static final String NOT_MATCHING_MSG_KEY = "java.collection.gpcoder.com.vn";

private Constant() {
super();
}
}

HeadersExchangeProducer.java

package com.gpcoder.exchange2exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import static com.gpcoder.exchange2exchange.Constant.*;

public class HeadersExchangeProducer {

private ExchangeChannel channel;

public void start() throws IOException, TimeoutException {
// Create connection
Connection connection = ConnectionManager.createConnection();

// Create channel
channel = new ExchangeChannel(connection, HEADERS_EXCHANGE_NAME);

// Create headers exchange
channel.declareExchange(BuiltinExchangeType.HEADERS);

// Create queues
channel.declareQueues(DEV_QUEUE_NAME, MANAGER_QUEUE_NAME, PUBLISHED_QUEUE_NAME);

// Binding queues with headers
Map<String, Object> devGroup = new HashMap<>();
devGroup.put("x-match", "any"); // Match any of the header
devGroup.put("dev", "Developer Channel");
devGroup.put("general", "General Channel");

Map<String, Object> managerGroup = new HashMap<>();
managerGroup.put("x-match", "any"); // Match any of the header
managerGroup.put("dev", "Developer Channel");
managerGroup.put("manager", "Manager Channel");
managerGroup.put("general", "General Channel");

Map<String, Object> publishedGroup = new HashMap<>();
publishedGroup.put("x-match", "all"); // Match all of the header
publishedGroup.put("general", "General Channel");
publishedGroup.put("access", "publish");

channel.performQueueBinding(DEV_QUEUE_NAME, "", devGroup);
channel.performQueueBinding(MANAGER_QUEUE_NAME, "", managerGroup);
channel.performQueueBinding(PUBLISHED_QUEUE_NAME, "", publishedGroup);
}
}

TopicExchangeProducer.java

package com.gpcoder.exchange2exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import static com.gpcoder.exchange2exchange.Constant.*;

public class TopicExchangeProducer {

private ExchangeChannel channel;

public void start() throws IOException, TimeoutException {
// Create connection
Connection connection = ConnectionManager.createConnection();

// Create channel
channel = new ExchangeChannel(connection, TOPIC_EXCHANGE_NAME);

// Create topic exchange
channel.declareExchange(BuiltinExchangeType.TOPIC);

// Create queues
channel.declareQueues(JAVA_QUEUE_NAME, ALL_QUEUE_NAME);

// Binding queues without headers
channel.performQueueBinding(DEV_QUEUE_NAME, JAVA_ROUTING_KEY, null);
channel.performQueueBinding(MANAGER_QUEUE_NAME, GPCODER_ROUTING_KEY, null);

// Binding headers exchange to topic exchange
channel.performExchangeBinding(HEADERS_EXCHANGE_NAME, TOPIC_EXCHANGE_NAME, GPCODER_ROUTING_KEY);
}

public void send(String message, String messageKey, Map<String, Object> headers) throws IOException {
// Send message
channel.publishMessage(message, messageKey, headers);
}
}

HeadersExchangeConsumer.java

package com.gpcoder.exchange2exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import static com.gpcoder.exchange2exchange.Constant.*;

public class HeadersExchangeConsumer {

private ExchangeChannel channel;

public void start() throws IOException, TimeoutException {
// Create connection
Connection connection = ConnectionManager.createConnection();

// Create channel
channel = new ExchangeChannel(connection, HEADERS_EXCHANGE_NAME);

// Create headers exchange
channel.declareExchange(BuiltinExchangeType.HEADERS);

// Create queues
channel.declareQueues(DEV_QUEUE_NAME, MANAGER_QUEUE_NAME, PUBLISHED_QUEUE_NAME);

// Binding queues with headers
Map<String, Object> devGroup = new HashMap<>();
devGroup.put("x-match", "any"); // Match any of the header
devGroup.put("dev", "Developer Channel");
devGroup.put("general", "General Channel");

Map<String, Object> managerGroup = new HashMap<>();
managerGroup.put("x-match", "any"); // Match any of the header
managerGroup.put("dev", "Developer Channel");
managerGroup.put("manager", "Manager Channel");
managerGroup.put("general", "General Channel");

Map<String, Object> publishedGroup = new HashMap<>();
publishedGroup.put("x-match", "all"); // Match all of the header
publishedGroup.put("general", "General Channel");
publishedGroup.put("access", "publish");

channel.performQueueBinding(DEV_QUEUE_NAME, "", devGroup);
channel.performQueueBinding(MANAGER_QUEUE_NAME, "", managerGroup);
channel.performQueueBinding(PUBLISHED_QUEUE_NAME, "", publishedGroup);
}

public void subscribe() throws IOException {
// Subscribe message
channel.subscribeMessage(DEV_QUEUE_NAME);
channel.subscribeMessage(MANAGER_QUEUE_NAME);
channel.subscribeMessage(PUBLISHED_QUEUE_NAME);
}
}

TopicExchangeConsumer.java

package com.gpcoder.exchange2exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import static com.gpcoder.exchange2exchange.Constant.*;
import static com.gpcoder.exchange2exchange.Constant.PUBLISHED_QUEUE_NAME;

public class TopicExchangeConsumer {

private ExchangeChannel channel;

public void start() throws IOException, TimeoutException {
// Create connection
Connection connection = ConnectionManager.createConnection();

// Create channel
channel = new ExchangeChannel(connection, TOPIC_EXCHANGE_NAME);

// Create topic exchange
channel.declareExchange(BuiltinExchangeType.TOPIC);

// Create queues
channel.declareQueues(JAVA_QUEUE_NAME, ALL_QUEUE_NAME);

// Binding queues without headers
channel.performQueueBinding(JAVA_QUEUE_NAME, JAVA_ROUTING_KEY, null);
channel.performQueueBinding(ALL_QUEUE_NAME, GPCODER_ROUTING_KEY, null);

// Binding headers exchange to topic exchange
channel.performExchangeBinding(HEADERS_EXCHANGE_NAME, TOPIC_EXCHANGE_NAME, GPCODER_ROUTING_KEY);
}

public void subscribe() throws IOException {
// Subscribe message
channel.subscribeMessage(JAVA_QUEUE_NAME);
channel.subscribeMessage(ALL_QUEUE_NAME);
}
}

App.java

package com.gpcoder.exchange2exchange;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import static com.gpcoder.exchange2exchange.Constant.DESIGN_PATTERN_MSG_KEY;

public class App {

public static void main(String[] args) throws IOException, TimeoutException {
HeadersExchangeProducer producer1 = new HeadersExchangeProducer();
producer1.start();

TopicExchangeProducer producer2 = new TopicExchangeProducer();
producer2.start();

// Publish some messages
Map<String, Object> devHeader = new HashMap<>();
devHeader.put("dev", "Developer Channel");
producer2.send("[1] Head First Design Pattern", DESIGN_PATTERN_MSG_KEY, devHeader);

HeadersExchangeConsumer consumer1 = new HeadersExchangeConsumer();
consumer1.start();
consumer1.subscribe();

TopicExchangeConsumer consumer2 = new TopicExchangeConsumer();
consumer2.start();
consumer2.subscribe();
}
}

Output của chương trình:

[Send] [{dev=Developer Channel}]: [1] Head First Design Pattern
[Received] [QDeveloper]: amq.ctag-QVREG0uBg6XsIebvbecqCg
[Received] [QDeveloper]: [1] Head First Design Pattern
[Received] [QManager]: amq.ctag-vLHleyPNedl2ZiMYukIing
[Received] [QManager]: [1] Head First Design Pattern
[Received] [QAll]: amq.ctag-GymeB0uwpALjgoM3RP_cvg
[Received] [QAll]: [1] Head First Design Pattern

Như bạn thấy, Message được gởi tới Exchange với routing key là design-pattern.gpcoder.com và header là {“dev”, “Developer Channel”} sẽ được chuyển tới Queue QAll và GPCoderHeadersExchange. Sau đó GPCoderHeadersExchange sẽ chuyển đến Queue QDeveloper và QManager dựa vào header {“dev”, “Developer Channel”}.

Tóm lại, Exchange Binding trong RabbitMQ là một tính năng vô cùng mạnh mẽ, giúp chúng ta kết hợp nhiều loại Exchange lại với nhau, giúp hệ thống mở rộng tốt hơn và đáp ứng được hầu hết yêu cầu phức tạp của hệ thống.

Bài viết gốc được đăng tải tại gpcoder.com

Có thể bạn quan tâm:

Xem thêm Việc làm Developer hấp dẫn trên TopDev

RELATED ARTICLESMORE FROM AUTHOR