RabbitMQ topic for Java【入门教程 5】

匹配交换器
通过匹配交换器，我们可以配置更灵活的消息系统，你可以在匹配交换器模式下发送这样的路由关键字：

“a.b.c”、“c.d”、“quick.orange.rabbit”

不过一定要记住，路由关键字【routingKey】不能超过255个字节（bytes）

匹配交换器的匹配符

*（星号）表示一个单词
#（井号）表示零个或者多个单词
示例说明：
这一章的例子中，我们使用三个段式的路由关键字，有三个单词和两个点组成。第一个词是速度，第二个词是颜色，第三个是动物名称。

我们用三个关键字来绑定，Q1绑定关键字是【*.orange.*】,Q2绑定关键字是【*.*.rabbit】和【lazy.#】，然后分析会发生什么：

RabbitMQ topic for Java【入门教程 5】

Q1会收到所有orange这种颜色相关的消息
Q2会收到所有rabbit这个动物相关的消息和所有速度lazy的动物的消息

分析：
生产者发送“quick.orange.rabbit”的消息，两个队列都会收到

生产者发送“lazy.orange.elephant”，两队列也都会收到。

生产者发送"quick.orange.fox"，那么只有Q1会收到。

生产者发送"lazy.brown.fox"，那么只会有Q2能收到。

生产者发送"quick.brown.fox"，那么这条消息会被丢弃，谁也收不到。

生产者发送"quick.orange.male.rabbit"，这个消息也会被丢弃，谁也收不到。

生产者发送"lazy.orange.male.rabbit"，这个消息会被Q2的【lazy.#】规则匹配上，发送到Q2队列中。

注意

交换器在匹配模式下：

如果消费者端的路由关键字只使用【#】来匹配消息，在匹配【topic】模式下，它会变成一个分发【fanout】模式，接收所有消息。

如果消费者端的路由关键字中没有【#】或者【*】，它就变成直连【direct】模式来工作。

代码如下：

RabbitMQ topic for Java【入门教程 5】

生产者：

package wxtest.rabbitMq.topic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import wxtest.rabbitMq.ConnectionUtil;

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

public class msgProduce {

    private static String EXCHAGE_NAME = "exchange_topic";//设置一个交换机
    private static final String[] routingKeys = new String[]{"info", "warning", "error"};
    
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHAGE_NAME, "direct");//fanout属于主题订阅方式 设置路由无效  direct和路由配合使用

        //分发消息
        for (String severity : routingKeys) {
            String message = "Send the message level:" + severity;
            channel.basicPublish(EXCHAGE_NAME, severity, null, message.getBytes());
            System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
        }

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

emailConsumer如下：

package wxtest.rabbitMq.topic;

import com.rabbitmq.client.*;
import wxtest.rabbitMq.ConnectionUtil;

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

public class emailConsumer {
    private static String exchage_name = "exchange_topic";//设置一个交换机
    private static final String[] routingKeys = new String[]{"info", "warning", "error"};

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(exchage_name, "direct");//绑定交换机
        String queueName = channel.queueDeclare().getQueue();
        for (String str : routingKeys) {
            channel.queueBind(queueName, exchage_name, str);//绑定交换机所对应的队列 只接受sms信息
        }


        System.out.println(" emailConsumer Waiting for messages. To exit press CTRL+C");
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println(" emailConsumer [x] Received '" + msg + "'");
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

smsConsumer如下：

package wxtest.rabbitMq.topic;

import com.rabbitmq.client.*;
import wxtest.rabbitMq.ConnectionUtil;

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

public class smsComsumer {
    private static String exchage_name = "exchange_topic";//设置一个交换机
    private static final String[] routingKeys = new String[]{"error"};

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(exchage_name, "direct");//绑定交换机
        String queueName = channel.queueDeclare().getQueue();
        for (String str : routingKeys) {
            channel.queueBind(queueName, exchage_name, str);//绑定交换机所对应的队列 只接受sms信息
        }
        System.out.println(" smsComsumer Waiting for messages. To exit press CTRL+C");
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println(" smsComsumer [x] Received '" + msg + "'");
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

输出结果如下：

RabbitMQ topic for Java【入门教程 5】

这种方式类似于模糊查询。

我们看到，队列1收到了所有的消息，队列2只收到了error级别的消息。这与我们的预期一样。

下一阶段我们可以进入RPC的学习了。

RabbitMQ topic for Java【入门教程 5】

注意

相关推荐