RabbitMQ（七）主题交换机

RabbitMQ官网中文版教程：

http://rabbitmq.mr-ping.com/tutorials_with_python/[5]Topics.html

上述教程示例为pathon版，Java版及相应解释如下：

生产者

package com.xc.rabbit.topic;

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

/**
 * Created by xc.
 */
public class EmitLogTopic {

    private static final String EXCHANGE_NAME = "top_exchange";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("rabbit");
        factory.setPassword("carrot");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String[] routingKeys = new String[] {
                "quick.orange.rabbit",
                "lazy.orange.elephant",
                "quick.orange.fox",
                "lazy.brown.fox",
                "quick.brown.fox",
                "quick.orange.male.rabbit",
                "lazy.orange.male.rabbit"
        };

        for (String routingKey : routingKeys) {
            String message = "From " + routingKey + " routingKey 's message";
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
            System.out.println("TopicSend [x] Sent '" + routingKey + "':'" + message + "'");
        }

        channel.close();
        connection.close();
    }
}

消费者1

package com.xc.topic;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created by xc.
 */
public class RecvLogsTopic1 {

    private static final String EXCHANGE_NAME = "top_exchange";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("rabbit");
        factory.setPassword("carrot");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String[] bindingKeys = new String[] {"*.orange.*"};

        String queueName = channel.queueDeclare().getQueue();

        for (String bindingKey : bindingKeys) {
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
            System.out.println("RecvLogsTopic1 exchange : " + EXCHANGE_NAME +
                    ", queue : " + queueName + ", BindRoutingKey : " + bindingKey);
        }

        System.out.println("RecvLogsTopic1 [*] Waiting for messages. To exit press CTRL + C");

        Consumer consumer = new DefaultConsumer(channel) {
            @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("RecvLogsTopic1 [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

消费者2

package com.xc.rabbit.topic;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created by xc.
 */
public class RecvLogsTopic2 {

    // 交换器名称
    private static final String EXCHANGE_NAME = "top_exchange";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("rabbit");
        factory.setPassword("carrot");
        // 创建一个新的连接
        Connection connection = factory.newConnection();
        // 创建一个频道
        Channel channel = connection.createChannel();
        // 声明一个匹配模式的交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        // 获取匿名队列名称
        String queueName = channel.queueDeclare().getQueue();
        // 路由关键字
        String[] routingKeys = new String [] {"*.*.rabbit", "lazy.#"};
        // 绑定路由关键字
        for (String bindingKey : routingKeys) {
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
            System.out.println("RecvLogsTopic2 exchange : " + EXCHANGE_NAME +
                    ", queue : " + queueName + ", BindRoutingKey : " + bindingKey);

        }
        System.out.println("RecvLogsTopic2 [*] Waiting for messages. To exit press CTRL + C");

        Consumer consumer = new DefaultConsumer(channel) {
            @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("RecvLogsTopic2 [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

先运行消费者，在运行生产者，结果如下：

不同路由模式的消息发送到了不同的队列。

注：

1）There can be as many words in the routing key as you like, up to the limit of 255 bytes.
routing key限长255字节以内

2）However there are two important special cases for binding keys:

* (star) can substitute for exactly one word.  用来表示一个单词
# (hash) can substitute for zero or more words.  用来表示零个或多个单词

3）When a queue is bound with "#" (hash) binding key - it will receive all the messages, regardless of the routing key - like in fanout exchange.
当一个queue的binding key 为 “#”， 不管routing key是什么，它会接收所有消息。

When special characters "" (star) and "#" (hash) aren't used in bindings, the topic exchange will behave just like a direct one.
在绑定时，未使用“”、“#”, 则topic交换器行为和direct交换器时相同的。

have fun

1. Will "*" binding catch a message sent with an empty routing key?
2. Will "#.*" catch a message with a string ".." as a key? Will it catch a message with a single word key?
3. How different is "a.*.#" from "a.#"?

*answer : *

1. 收不到。
2. 可以。 可以。
3. 就如同 * 和 # 的区别。