architecture - 发布订阅模型与主题交换

Question

我正在开发具有事件驱动架构的缓存服务器，它将按如下方式工作:

我想将 Set 操作发送到所有副本(扇出交换(？))并将 Get 发送到任意一个(默认交换(？))。

我已阅读 Publish&Subscribe模式，并能够使用 fanout exchange 使所有服务器响应。我读过 RPC模型并能够做出任意的服务器响应。但我无法将这些方法统一到一个架构中。请帮忙。

问题:

1. 组织 MQ 以实现此行为的最佳方式是什么？
2. 我应该将两个队列绑定(bind)到一个交换器中吗？
3. 我想使用 correlationId 从服务器响应客户端。我应该重用现有的队列/交换还是创建新的？

Answer 1

在通过你的问题领域后，我的理解是 - 在运行时，多个客户端将发送“set”和“get”消息到RabbitMQ和每个“设置” 消息由当时处于事件状态的每个服务器缓存处理。并且 “get” 消息需要由任何一个服务器缓存处理，并且需要将 response 消息发送回发送 “get” 的客户端 消息。

如果我错了，请纠正我。

在这种情况下，可以公平地假设在客户端将有单独的触发点来生成/发布"get"/"set" 消息。因此，逻辑上“获取”消息生产者和“设置”消息发布者将是两个独立的程序/类。

因此，您对发布/订阅和 RPC 模型的选择看起来合乎逻辑。您唯一需要做的就是将 “set” 和 “get” 消息处理与服务器缓存结合起来，使用两个单独的 channel (在相同的连接)服务器缓存中的每个 set 和 get 消息一个。 引用我在下面附上的代码。我使用了您在问题中提到的相同示例(来自 rabbitmq 站点)的 java 代码。一些小的修改，它非常简单。在 python 中做同样的事情也不难。

现在向你提问 -

What is the best way to organize MQ to achieve this behavior?

您对发布/订阅和 RPC 模型的选择看起来合乎逻辑。客户端将向交换器发布“set”消息(类型Fanout，例如名称“set_ex”)并且每个服务器缓存实例将被监听到他们的临时队列(持续到连接生效)，这些队列将被绑定(bind)到交换“set_ex”。客户端将生成“get”消息到交换器(类型Direct，例如名称“get_ex”)和队列“get_q” 将与其队列名称绑定(bind)到此交换。每个服务器缓存都将监听此“get_q”。服务器缓存会将结果消息发送到与“get” 消息一起传递的临时队列名称。一旦客户端收到response 消息，连接就会关闭，临时队列也会被移除。 (注意 - 在下面的示例代码中，我在默认交换中绑定(bind)了“get_q”，就像 rabbitmq 站点上的示例一样。但是将“get_q”绑定(bind)到单独的交换(直接类型)并不困难以获得更好的可管理性。)

Should I bind two queues into one exchange?

我认为这不是一个正确的选择，因为对于发布/订阅场景，您将明确需要一个扇出交换器，并且发送到扇出交换器的每条消息都会被复制到绑定(bind)到该交换器的每个队列。而且我们不希望 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” 消息一起传递，正如在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();            
                }
            }
        }

    }

}

希望这对您有所帮助。