Rabbitmq打怪升级之路（十一）Topic-主题通配符模式

简书：亚武de小文【原创：转载请注明出处】

主题交换机模式（Topic）

通配符工作模式

LengToo上学.png

RabbitMQ有以下几种工作模式：

Work queues
Publish/Subscribe
Routing
Topic
Headers
RPC

Topic

模型图

[亚武de小文]Topic模型图.png

Producer：生产者，是发消息方。
Exchanger：路由器交换机，此处类型为topic，它会把消息的路由键取出来，与绑定它的队列做路由键匹配。如果队列关心的路由键能匹配上消息的路由键，则将这个消息投递到这个队列中。
Queue01：一个队列，它是关心orange颜色的队列。
Queue02：一个队列，它是关心rabbit与lazy的队列。
Consumer01：消费者，从Queue01中获取消息。
Consumer01：消费者，从Queue02中获取消息。
* 可以替代一个单词
# 可以替换零个或多个单词

主题交换机模式：
1、每个消费者监听自己的队列，并且设置带通配符的routingkey。
2、生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。

参考代码

生产者

队列绑定交换机指定通配符：【统配符规则】 中间以“.”分隔。符号#可以匹配多个词，符号*可以匹配一个词语。
声明交换机，指定topic类型：

Producer.java

package com.yawu.xiaowen.topic;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Topic交换机
 * 生产者
 *
 * @author yawu
 * @date 2019.06.30
 */
public class Producer {
    private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
    private static final String EXCHANGE_NAME = "mq_topic_exchange";

    public static void main(String[] args) {
        try {
            // RabbitMQ建立连接的管理器
            ConnectionFactory factory = new ConnectionFactory();
            // 设置服务器地址
            factory.setHost("127.0.0.1");
            factory.setUsername("guest");
            factory.setPassword("guest");

            // 创建一个连接
            Connection connection = factory.newConnection();
            // 创建一个信道
            Channel channel = connection.createChannel();

            String message = "开始发送信息-topic交换机";

            //声明一个TOPIC类型的交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

            // 向交换机发送消息
            // 第一种情况：发送【*.orange.*】通配符消息
            channel.basicPublish(EXCHANGE_NAME, "like.orange.color", null, message.getBytes("UTF-8"));
            // 第二种情况：发送【lazy.#】通配符消息
//            channel.basicPublish(EXCHANGE_NAME, "lazy.boy.girl", null, message.getBytes("UTF-8"));

            LOGGER.info("消息发送成功:{}", message);
            channel.close();
            connection.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

消费者

Consumer01.java

package com.yawu.xiaowen.topic;

import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Topic交换机
 * 消费者01
 * @date 2019.06.30
 * @author yawu
 */
public class Consumer01 {
    private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
    private static final String EXCHANGE_NAME = "mq_topic_exchange";
    private static final String QUEUE_NAME_01 = "mq_topic_queue_01";
    private static final String QUEUE_NAME_02 = "mq_topic_queue_02";

    public static void main(String[] args) {
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            //声明一个Topic类型的交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

            // 声明两个队列
            /**
             * 【参数详解】
             * queue：要创建的队列名
             * durable：是否持久化。如果为true，可以在RabbitMQ崩溃后恢复消息
             * exclusive：true表示一个队列只能被一个消费者占有并消费
             * autoDelete：true表示服务器不在使用这个队列是会自动删除它
             * arguments：其它参数
             */
            channel.queueDeclare(QUEUE_NAME_01, true, false, false, null);
            channel.queueDeclare(QUEUE_NAME_02, true, false, false, null);

            final String ROUTING_KEY_ORANGE = "*.orange.*";
            final String ROUTING_KEY_LAZY = "lazy.#";
            // 队列一对ORANGE感兴趣，匹配  XXX.orange.XXX 的消息
            channel.queueBind(QUEUE_NAME_01, EXCHANGE_NAME, ROUTING_KEY_ORANGE);
            // 队列二对LAZY感兴趣，匹配  lazy.XXX.XXX.XXX
            channel.queueBind(QUEUE_NAME_02, EXCHANGE_NAME, ROUTING_KEY_LAZY);

            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    final String message = new String(body, "UTF-8");
                    LOGGER.info("队列一收到消息:{}", message);
                }
            };

            // 队列一确认消息
            channel.basicConsume(QUEUE_NAME_01, true, consumer);

        } catch (Exception e) {
            LOGGER.error("an exception was occurred , caused by :{}", e.getMessage());
        }
    }
}

Consumer02.java

package com.yawu.xiaowen.topic;

import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Topic交换机
 * 消费者02
 * @date 2019.06.30
 * @author yawu
 */
public class Consumer02 {
    private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
    private static final String EXCHANGE_NAME = "mq_topic_exchange";
    private static final String QUEUE_NAME_01 = "mq_topic_queue_01";
    private static final String QUEUE_NAME_02 = "mq_topic_queue_02";

    public static void main(String[] args) {
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            //声明一个Topic类型的交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

            // 声明两个队列
            /**
             * 【参数详解】
             * queue：要创建的队列名
             * durable：是否持久化。如果为true，可以在RabbitMQ崩溃后恢复消息
             * exclusive：true表示一个队列只能被一个消费者占有并消费
             * autoDelete：true表示服务器不在使用这个队列是会自动删除它
             * arguments：其它参数
             */
            channel.queueDeclare(QUEUE_NAME_01, true, false, false, null);
            channel.queueDeclare(QUEUE_NAME_02, true, false, false, null);

            final String ROUTING_KEY_ORANGE = "*.orange.*";
            final String ROUTING_KEY_LAZY = "lazy.#";
            // 队列一对ORANGE感兴趣，匹配  XXX.orange.XXX 的消息
            channel.queueBind(QUEUE_NAME_01, EXCHANGE_NAME, ROUTING_KEY_ORANGE);
            // 队列二对LAZY感兴趣，匹配  lazy.XXX.XXX.XXX
            channel.queueBind(QUEUE_NAME_02, EXCHANGE_NAME, ROUTING_KEY_LAZY);

            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    final String message = new String(body, "UTF-8");
                    LOGGER.info("队列二收到消息:{}", message);
                }
            };

            //队列一确认消息
            channel.basicConsume(QUEUE_NAME_02, true, consumer);

        } catch (Exception e) {
            LOGGER.error("an exception was occurred , caused by :{}", e.getMessage());
        }
    }
}