发布订阅模型与主题交换
Publish&Subscribe model together with topic exchange
我正在开发具有事件驱动架构的缓存服务器，它将按如下方式工作：
我想 Set 操作发送到所有副本（扇出交换（？））和 Get 发送到任意一个副本（默认交换（？））。
我已经阅读了 Publish&Subscribe pattern and was able to make all servers response using fanout exchange. I've read about RPC 模型并且能够做出任意服务器响应。但我无法将这些方法统一到一个架构中。请帮忙。
问题：
- 组织 MQ 以实现此行为的最佳方式是什么？
- 是否应该将两个队列绑定到一个交换器中？
- 我想用
correlationId 从服务器响应客户端。我应该重复使用现有的 queues/exchanges 还是创建新的？
您需要一个 topic 队列用于您的集合操作。这样 N 个客户端可以 receive/react 到消息。
通过您的问题域后，我的理解是 - 在 运行 时间，多个客户端将发送 "set" 和 "get" 消息到 RabbitMQ，每个 "set" 消息将由当时活动的每个服务器缓存处理。并且 "get" 消息需要由任何一个服务器缓存处理，并且需要将 response 消息发送回客户端发送了 "get" 消息。
如有错误请指正
在这种情况下，可以公平地假设 producing/publishing "get"/"set"[=71= 的客户端会有单独的触发点] 消息。因此，逻辑上 "get" 消息生产者和 "set" 消息发布者将是两个独立的 programs/classes.
因此，您选择 pub/sub 和 RPC 模型看起来合乎逻辑。您唯一需要做的就是将 "set" 和 "get" 消息处理和 Server Cache 联合起来，这可以使用两个单独的通道（在同一连接内）很容易完成服务器缓存上的每个 set 和 get 消息。 参考我的代码附在下面。我使用了您在问题中提到的相同样本（来自 rabbitmq 站点）的 java 代码。一些小的修改，它非常简单。在 python 中做同样的事情也不难。
现在向您提问 -
What is the best way to organize MQ to achieve this behavior?
您选择的 pub/sub 和 RPC 模型看起来合乎逻辑。
客户端将发布 "set" 消息到交换器（类型 Fanout，例如名称 "set_ex") 并且每个服务器缓存实例将监听它们的临时队列（持续到连接有效），这些队列将被绑定到交换 "set_ex"。
客户端将向交换器生成 "get" 消息（类型 Direct，例如名称 "get_ex") 和队列 "get_q" 将使用其队列名称绑定到此交换。每个服务器缓存都将监听此 "get_q"。服务器缓存会将结果消息发送到与 "get" 消息一起传递的临时队列名称。客户端收到 response 消息后，连接将关闭并删除临时队列。 （注意 - 在下面的示例代码中，我在默认交换中绑定了 "get_q" ，就像 rabbitmq 站点上的示例一样。但是将 "get_q" 绑定到一个单独交换（直接类型）以获得更好的可管理性。）
Should I bind two queues into one exchange?
我认为这不是一个正确的选择，因为对于 pub/sub 场景，您将明确需要一个扇出交换器，并且发送到扇出交换器的每条消息都会被复制到绑定到该交换器的每个队列。而且我们不希望get消息被推送到所有Server Cache。
I would like to response from Server to client with correlationId.
Should I reuse existing queues/exchanges or create new one?
您需要做的就是将响应消息从服务器发送到 tempQueueName，并与原始 "get"[=71 一起传递=] 消息，如 rabbitmq 提供的示例中所使用的那样。
发布"set"消息的客户端代码。
public class Client {
private static final String EXCHANGE_NAME_SET = "set_ex";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME_SET, BuiltinExchangeType.FANOUT);
String message = getMessage(args);
channel.basicPublish(EXCHANGE_NAME_SET, "", null, message.getBytes("UTF-8"));
System.out.println("Sent '" + message + "'");
channel.close();
connection.close();
}
private static String getMessage(String[] strings) {
if (strings.length < 1)
return "info: Hello World!";
return joinStrings(strings, " ");
}
private static String joinStrings(String[] strings, String delimiter) {
int length = strings.length;
if (length == 0)
return "";
StringBuilder words = new StringBuilder(strings[0]);
for (int i = 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
用于生成 "get" 消息并接收响应消息的客户端代码。
public class RPCClient {
private static final String EXCHANGE_NAME_GET = "get_ex";
private Connection connection;
private Channel channel;
private String requestQueueName = "get_q";
private String replyQueueName;
public RPCClient() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
replyQueueName = channel.queueDeclare().getQueue();
}
public String call(String message) throws IOException, InterruptedException {
String corrId = UUID.randomUUID().toString();
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();
//channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
}
}
});
return response.take();
}
public void close() throws IOException {
connection.close();
}
public static void main(String[] args) throws IOException, TimeoutException {
RPCClient rpcClient = null;
String response = null;
try {
rpcClient = new RPCClient();
System.out.println(" sending get message");
response = rpcClient.call("30");
System.out.println(" Got '" + response + "'");
}
catch (IOException | TimeoutException | InterruptedException e) {
e.printStackTrace();
}
finally {
if (rpcClient!= null) {
try {
rpcClient.close();
}
catch (IOException _ignore) {}
}
}
}
}
订阅"set"条消息并消费"get"条消息的服务器代码。
public class ServerCache1 {
private static final String EXCHANGE_NAME_SET = "set_ex";
private static final String EXCHANGE_NAME_GET = "get_ex";
private static final String RPC_GET_QUEUE_NAME = "get_q";
private static final String s = UUID.randomUUID().toString();
public static void main(String[] args) throws IOException, TimeoutException {
System.out.println("Server Id " + s);
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
// set server to receive and process set messages
Channel channelSet = connection.createChannel();
channelSet.exchangeDeclare(EXCHANGE_NAME_SET, BuiltinExchangeType.FANOUT);
String queueName = channelSet.queueDeclare().getQueue();
channelSet.queueBind(queueName, EXCHANGE_NAME_SET, "");
System.out.println("waiting for set message");
Consumer consumerSet = new DefaultConsumer(channelSet) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received '" + message + "'");
}
};
channelSet.basicConsume(queueName, true, consumerSet);
// here onwards following code is to set up Get message processing at Server cache
Channel channelGet = connection.createChannel();
channelGet.queueDeclare(RPC_GET_QUEUE_NAME, false, false, false, null);
channelGet.basicQos(1);
System.out.println("waiting for get message");
Consumer consumerGet = new DefaultConsumer(channelGet) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
AMQP.BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(properties.getCorrelationId())
.build();
System.out.println("received get message");
String response = "get response from server " + s;
channelGet.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
channelGet.basicAck(envelope.getDeliveryTag(), false);
// RabbitMq consumer worker thread notifies the RPC server owner thread
synchronized(this) {
this.notify();
}
}
};
channelGet.basicConsume(RPC_GET_QUEUE_NAME, false, consumerGet);
// Wait and be prepared to consume the message from RPC client.
while (true) {
synchronized(consumerGet) {
try {
consumerGet.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
希望对您有所帮助。
我正在开发具有事件驱动架构的缓存服务器，它将按如下方式工作：
我想 Set 操作发送到所有副本（扇出交换（？））和 Get 发送到任意一个副本（默认交换（？））。
我已经阅读了 Publish&Subscribe pattern and was able to make all servers response using fanout exchange. I've read about RPC 模型并且能够做出任意服务器响应。但我无法将这些方法统一到一个架构中。请帮忙。
问题：
- 组织 MQ 以实现此行为的最佳方式是什么？
- 是否应该将两个队列绑定到一个交换器中？
- 我想用
correlationId从服务器响应客户端。我应该重复使用现有的 queues/exchanges 还是创建新的？
您需要一个 topic 队列用于您的集合操作。这样 N 个客户端可以 receive/react 到消息。
通过您的问题域后，我的理解是 - 在 运行 时间，多个客户端将发送 "set" 和 "get" 消息到 RabbitMQ，每个 "set" 消息将由当时活动的每个服务器缓存处理。并且 "get" 消息需要由任何一个服务器缓存处理，并且需要将 response 消息发送回客户端发送了 "get" 消息。
如有错误请指正
在这种情况下，可以公平地假设 producing/publishing "get"/"set"[=71= 的客户端会有单独的触发点] 消息。因此，逻辑上 "get" 消息生产者和 "set" 消息发布者将是两个独立的 programs/classes.
因此，您选择 pub/sub 和 RPC 模型看起来合乎逻辑。您唯一需要做的就是将 "set" 和 "get" 消息处理和 Server Cache 联合起来，这可以使用两个单独的通道（在同一连接内）很容易完成服务器缓存上的每个 set 和 get 消息。 参考我的代码附在下面。我使用了您在问题中提到的相同样本（来自 rabbitmq 站点）的 java 代码。一些小的修改，它非常简单。在 python 中做同样的事情也不难。
现在向您提问 -
What is the best way to organize MQ to achieve this behavior?
您选择的 pub/sub 和 RPC 模型看起来合乎逻辑。 客户端将发布 "set" 消息到交换器（类型 Fanout，例如名称 "set_ex") 并且每个服务器缓存实例将监听它们的临时队列（持续到连接有效），这些队列将被绑定到交换 "set_ex"。 客户端将向交换器生成 "get" 消息（类型 Direct，例如名称 "get_ex") 和队列 "get_q" 将使用其队列名称绑定到此交换。每个服务器缓存都将监听此 "get_q"。服务器缓存会将结果消息发送到与 "get" 消息一起传递的临时队列名称。客户端收到 response 消息后，连接将关闭并删除临时队列。 （注意 - 在下面的示例代码中，我在默认交换中绑定了 "get_q" ，就像 rabbitmq 站点上的示例一样。但是将 "get_q" 绑定到一个单独交换（直接类型）以获得更好的可管理性。）
Should I bind two queues into one exchange?
我认为这不是一个正确的选择，因为对于 pub/sub 场景，您将明确需要一个扇出交换器，并且发送到扇出交换器的每条消息都会被复制到绑定到该交换器的每个队列。而且我们不希望get消息被推送到所有Server Cache。
I would like to response from Server to client with correlationId. Should I reuse existing queues/exchanges or create new one?
您需要做的就是将响应消息从服务器发送到 tempQueueName，并与原始 "get"[=71 一起传递=] 消息，如 rabbitmq 提供的示例中所使用的那样。
发布"set"消息的客户端代码。
public class Client {
private static final String EXCHANGE_NAME_SET = "set_ex";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME_SET, BuiltinExchangeType.FANOUT);
String message = getMessage(args);
channel.basicPublish(EXCHANGE_NAME_SET, "", null, message.getBytes("UTF-8"));
System.out.println("Sent '" + message + "'");
channel.close();
connection.close();
}
private static String getMessage(String[] strings) {
if (strings.length < 1)
return "info: Hello World!";
return joinStrings(strings, " ");
}
private static String joinStrings(String[] strings, String delimiter) {
int length = strings.length;
if (length == 0)
return "";
StringBuilder words = new StringBuilder(strings[0]);
for (int i = 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
用于生成 "get" 消息并接收响应消息的客户端代码。
public class RPCClient {
private static final String EXCHANGE_NAME_GET = "get_ex";
private Connection connection;
private Channel channel;
private String requestQueueName = "get_q";
private String replyQueueName;
public RPCClient() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
replyQueueName = channel.queueDeclare().getQueue();
}
public String call(String message) throws IOException, InterruptedException {
String corrId = UUID.randomUUID().toString();
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();
//channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
}
}
});
return response.take();
}
public void close() throws IOException {
connection.close();
}
public static void main(String[] args) throws IOException, TimeoutException {
RPCClient rpcClient = null;
String response = null;
try {
rpcClient = new RPCClient();
System.out.println(" sending get message");
response = rpcClient.call("30");
System.out.println(" Got '" + response + "'");
}
catch (IOException | TimeoutException | InterruptedException e) {
e.printStackTrace();
}
finally {
if (rpcClient!= null) {
try {
rpcClient.close();
}
catch (IOException _ignore) {}
}
}
}
}
订阅"set"条消息并消费"get"条消息的服务器代码。
public class ServerCache1 {
private static final String EXCHANGE_NAME_SET = "set_ex";
private static final String EXCHANGE_NAME_GET = "get_ex";
private static final String RPC_GET_QUEUE_NAME = "get_q";
private static final String s = UUID.randomUUID().toString();
public static void main(String[] args) throws IOException, TimeoutException {
System.out.println("Server Id " + s);
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
// set server to receive and process set messages
Channel channelSet = connection.createChannel();
channelSet.exchangeDeclare(EXCHANGE_NAME_SET, BuiltinExchangeType.FANOUT);
String queueName = channelSet.queueDeclare().getQueue();
channelSet.queueBind(queueName, EXCHANGE_NAME_SET, "");
System.out.println("waiting for set message");
Consumer consumerSet = new DefaultConsumer(channelSet) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received '" + message + "'");
}
};
channelSet.basicConsume(queueName, true, consumerSet);
// here onwards following code is to set up Get message processing at Server cache
Channel channelGet = connection.createChannel();
channelGet.queueDeclare(RPC_GET_QUEUE_NAME, false, false, false, null);
channelGet.basicQos(1);
System.out.println("waiting for get message");
Consumer consumerGet = new DefaultConsumer(channelGet) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
AMQP.BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(properties.getCorrelationId())
.build();
System.out.println("received get message");
String response = "get response from server " + s;
channelGet.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
channelGet.basicAck(envelope.getDeliveryTag(), false);
// RabbitMq consumer worker thread notifies the RPC server owner thread
synchronized(this) {
this.notify();
}
}
};
channelGet.basicConsume(RPC_GET_QUEUE_NAME, false, consumerGet);
// Wait and be prepared to consume the message from RPC client.
while (true) {
synchronized(consumerGet) {
try {
consumerGet.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
希望对您有所帮助。