RabbitMQ provides 6 message models, but the first 5 are commonly used, and the sixth is actually RPC, so generally speaking, you can understand the first 5, and the latter three are divided according to the Exchange type.
Note: The explanation of the following modes is mainly based on Java native API operations , so the following dependencies need to be added to the project.
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.9.0</version></dependency>For subsequent operations, first define a connection tool class that connects to rabbitmq
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQUtils {
private static ConnectionFactory connectionFactory;
static { connectionFactory = new ConnectionFactory(); //我们把重量级资源通过单例模式加载 connectionFactory.setHost("192.168.153.128"); connectionFactory.setPort(5672); connectionFactory.setUsername("admin"); connectionFactory.setPassword("admin"); //上面创建的VHost connectionFactory.setVirtualHost("/order"); }
//定义提供连接对象的方法 public static Connection getConnection() { try { return connectionFactory.newConnection(); } catch (Exception e) { e.printStackTrace(); } return null; }
//定义关闭通道和关闭连接工具方法 public static void closeConnectionAndChanel(Channel channel, Connection conn) { try { if (channel != null) { channel.close(); } if (conn != null) { conn.close(); } } catch (Exception e) { e.printStackTrace(); } }}Basic message model
RabbitMQ is a message broker: it accepts and forwards messages. Think of it like a post office: when you put mail for delivery in a mailbox, you can be sure that the letter carrier will eventually deliver the mail to your recipient. In this analogy, RabbitMQ is a mailbox, a post office, and a letter carrier.
- P: Producer, send message to message queue
- C: Consumer: The receiver of the message, will always wait for the message to arrive.
- queue: message queue, the red part in the figure. Similar to a mailbox, messages can be cached; producers deliver messages to it, and consumers get messages from it.
1. Send a message
In the native Java API, the queueDeclarequeue is declared through the method:
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) throws IOException;Parameter Description:
- queue , the queue name.
- durable , whether it is persistent, if persistent, the queue is still there after mq restarts.
- exclusive , whether the connection is exclusive, the queue is only allowed to access in this connection, if the connection connection closes the queue, it will be automatically deleted, if this parameter is set to true, it can be used to create a temporary queue.
- autoDelete , automatically delete, whether the queue is automatically deleted when the queue is no longer used, if this parameter and the exclusive parameter are set to true, a temporary queue can be implemented (the queue will be automatically deleted when it is not used).
- The arguments parameter, you can set the extended parameters of a queue, such as: you can set the survival time, etc.
Mainly by basicPublishmethod
void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;Parameter Description:
- exchange , the exchange, if not specified will use mq’s default exchange (set to “”).
- routingKey , routing key, the switch forwards the message to the specified queue according to the routing key. If the default switch is used, routingKey is set to the name of the queue.
- props , the properties of the message.
- body , the message content.
Code
Producer
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;
public class Producer {
//定义队列名称 private final static String QUEUE_NAME = "hello";
public static void main(String[] argv) throws Exception { // 1、获取到连接 Connection connection = RabbitMQUtils.getConnection(); // 2、从连接中创建通道，使用通道才能完成消息相关的操作 Channel channel = connection.createChannel(); // 3、声明（创建）队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 4、消息内容 String message = "Hello World!"; // 向指定的队列中发送消息 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); //关闭通道和连接 channel.close(); connection.close(); }}Go to the console to view:
2. Receive messages
Receive message consumer#handleDeliverymethod:
void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException;Consumer
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
public class Consumer { private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = RabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //实现消费方法 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { System.out.println(new String(body)); } }; //自动ack channel.basicConsume(QUEUE_NAME, true, consumer); }}work message model
Multiple consumers listen to the same queue. After the consumer receives the message, it consumes it asynchronously through the thread pool. But a message can only be fetched by one consumer. Work queues are often used to avoid message accumulation problems.
- P: Producer, publish task.
- C1: Consumer 1, receives the task and completes the task, assuming that the completion speed is slow (simulation takes time)
- C2: Consumer 2, receives the task and completes the task, assuming the completion speed is faster
Producer
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQUtils {
//定义提供连接对象的方法 public static Connection getConnection() { try { ConnectionFactory connectionFactory = new ConnectionFactory(); //我们把重量级资源通过单例模式加载 connectionFactory.setHost("192.168.153.128"); connectionFactory.setPort(5672); connectionFactory.setUsername("admin"); connectionFactory.setPassword("admin"); connectionFactory.setVirtualHost("order"); return connectionFactory.newConnection(); } catch (Exception e) { e.printStackTrace(); } return null; }
//定义关闭通道和关闭连接工具方法 public static void closeConnectionAndChanel(Channel channel, Connection conn) { try { if (channel != null) { channel.close(); } if (conn != null) { conn.close(); } } catch (Exception e) { e.printStackTrace(); } }}Consumer1
import cn.javatv.javaAPI.RabbitMQUtils;import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
import java.util.concurrent.TimeUnit;
public class Consumer1 { private final static String QUEUE_NAME = "work";
public static void main(String[] args) throws Exception { // 获取到连接 Connection connection = RabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //实现消费方法 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { try { //模拟任务耗时 TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Consumer1_" +new String(body)); } }; //自动ack channel.basicConsume(QUEUE_NAME, true, consumer); }}Consumer2
import cn.javatv.javaAPI.RabbitMQUtils;import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
public class Consumer2 { private final static String QUEUE_NAME = "work";
public static void main(String[] args) throws Exception { // 获取到连接 Connection connection = RabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //实现消费方法 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { System.out.println("Consumer2_" + new String(body)); } }; //自动ack channel.basicConsume(QUEUE_NAME, true, consumer); }}Start the consumer first, then start the generator, the output is as follows:
We found that consumers consume according to polling, but there is a problem with this kind of consumption. If the processing power of Consumer1 is extremely fast, the processing power of Consumer2 (which sleeps for 2s in the code) is extremely slow, which is that Consumer2 will seriously drag down the overall consumption progress, and Consuemr1 finished the task early and had nothing to do.
Able people should do more work
It can be seen from the above results that the tasks are distributed evenly, that is to say, no matter whether your last task is completed or not, I will continue to distribute the following tasks to you, but in fact, for efficiency, whoever consumes the faster, whoever get more. Therefore, the parameter of the BasicQos method can be set to 1, and the premise is that it will take effect only in the case of manual ack, that is, autoAck = false .
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
import java.io.IOException;import java.util.concurrent.TimeUnit;
public class Consumer2 { private final static String QUEUE_NAME = "work";
public static void main(String[] args) throws Exception { // 获取到连接 Connection connection = RabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //设置消费者同时只能处理一条消息 channel.basicQos(1); //实现消费方法 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try { //模拟任务耗时 TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Consumer2_" + new String(body)); //确认消息 channel.basicAck(envelope.getDeliveryTag(),false); } }; //手动ack channel.basicConsume(QUEUE_NAME, false, consumer); }}Output result:
It can be seen that Consumer1 consumes 19, and Consumer2 only consumes 1.
Publish/Subscribe-Fanout
To send messages to multiple consumers at once, the switch type for this mode is Fanout, also known as broadcast.
It has the following properties:
- There can be multiple consumers.
- Each consumer has its own queue.
- Each queue is bound to Exchange.
- The message sent by the producer can only be sent to the switch. The switch decides which queue to send to, but the producer cannot decide .
- The exchange sends the message to all queues to which it is bound.
- The consumers of the queue can get the message, so that a message can be consumed by multiple consumers .
Producer
import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;
import java.io.IOException;import java.util.concurrent.TimeoutException;
public class FanoutProducer {
public final static String EXCHANGE_NAME = "fanout";
public static void main(String[] args) throws IOException, TimeoutException { //建立连接 Connection connection = RabbitMQUtils.getConnection(); // 创建一个信道 Channel channel = connection.createChannel(); // 指定转发类型为FANOUT channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); //发送3条消息，且路由键不同 for (int i = 1; i <= 3; i++) { //路由键，循环3次，路由键为routekey1，routekey2，routekey3 String routekey = "routekey" + i; // 发送的消息 String message = "fanout_" + i; /* * 参数1：exchange name 交换机 * 参数2：routing key 路由键 */ channel.basicPublish(EXCHANGE_NAME, routekey, null, message.getBytes()); System.out.println(" [x] Sent '" + routekey + "':'" + message + "'"); } // 关闭 channel.close(); connection.close(); }}Consumer1
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
import java.io.IOException;
public class Consumer1 {
public final static String EXCHANGE_NAME = "fanout";
public static void main(String[] argv) throws IOException { Connection connection = RabbitMQUtils.getConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 声明一个随机队列 String queueName = channel.queueDeclare().getQueue(); /* * 队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定 */ String[] routekeys = {"routekey1", "routekey2", "routekey3"}; for (String routekey : routekeys) { //绑定 channel.queueBind(queueName, FanoutProducer.EXCHANGE_NAME, routekey); } System.out.println("[" + queueName + "]等待消息："); // 创建队列消费者 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("接收" + envelope.getRoutingKey() + "：" + message); } }; channel.basicConsume(queueName, true, consumer); }}Let’s look at the definition of fanout:
The message is broadcast to the bound queue, no matter what routing key is bound to the queue, the message passes through the exchange, and each queue has a copy.
In other words, as long as the queue is bound to the exchange, it can receive messages regardless of the routing key.
Such as binding a non-existing routing key:
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
import java.io.IOException;import java.util.concurrent.TimeoutException;
/** * 类说明：fanout消费者--绑定一个不存在的路由键 */public class Consumer2 {
public final static String EXCHANGE_NAME = "fanout";
public static void main(String[] argv) throws IOException, TimeoutException { Connection connection = RabbitMQUtils.getConnection(); final Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 声明一个随机队列 String queueName = channel.queueDeclare().getQueue(); //设置一个不存在的路由键 String routekey = "xxx"; channel.queueBind(queueName, FanoutProducer.EXCHANGE_NAME, routekey); System.out.println("队列[" + queueName + "]等待消息：");
// 创建队列消费者 final Consumer consumerB = 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("接收消息 [" + envelope.getRoutingKey() + "] " + message); } }; channel.basicConsume(queueName, true, consumerB); }}output:
队列[amq.gen-G2LL566wrSH3mGBUF6XKCQ]等待消息：接收消息 [routekey1] fanout_1接收消息 [routekey2] fanout_2接收消息 [routekey3] fanout_3No matter how we adjust the routing keys of producers and consumers, it has no effect on the receipt of messages.
Routing-Direct
In the Direct model, the binding between the queue and the exchange cannot be arbitrary, but a RoutingKey (routing key) must be specified. The sender of the message must also specify the routing key of the message when sending a message to the Exchange.
- P: Producer, sends a message to Exchange. When sending a message, a routing key is specified.
- X: Exchange, receives the message from the producer, and then delivers the message to the queue that exactly matches the routing key.
- C1: Consumer, whose queue specifies a message that requires the routing key to be error.
- C2: Consumer, whose queue specifies messages that require routing keys as info, error, and warning.
Producer
Send 3 different types of logs.
import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;
import java.io.IOException;import java.util.concurrent.TimeoutException;
public class DirectProducer {
public final static String EXCHANGE_NAME = "direct";
public static void main(String[] args) throws IOException, TimeoutException { //创建连接、连接到RabbitMQ Connection connection = RabbitMQUtils.getConnection(); //创建信道 Channel channel = connection.createChannel(); //在信道中设置交换器 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //申明队列（放在消费者中去做） String[] routeKeys = {"info", "warning", "error"}; for (int i = 1; i <= 6; i++) { String routeKey = routeKeys[i % 3]; String msg = routeKey + "日志"; //发布消息 channel.basicPublish(EXCHANGE_NAME, routeKey, null, msg.getBytes()); System.out.println("Sent:" + msg); } channel.close(); connection.close(); }}Production message:
Sent:warning日志Sent:error日志Sent:info日志Sent:warning日志Sent:error日志Sent:info日志Consumer1
Specifies the message that requires the routing key to be error.
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
import java.io.IOException;import java.util.concurrent.TimeoutException;
public class Consumer1 {
public final static String EXCHANGE_NAME = "direct";
public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { //创建连接、连接到RabbitMQ Connection connection = RabbitMQUtils.getConnection(); //创建一个信道 final Channel channel = connection.createChannel(); //信道设置交换器类型(direct) channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //声明一个随机队列 String queueName = channel.queueDeclare().getQueue(); //绑定 channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, "error"); System.out.println("队列[" + queueName + "]等待消息："); // 创建队列消费者 final 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("接收消息 [" + envelope.getRoutingKey() + "] " + message); } }; channel.basicConsume(queueName, true, consumer); }}Receive message:
队列[amq.gen-NhIiesUDi547ZGr4JBEsnA]等待消息：接收消息 [error] error日志接收消息 [error] error日志Consumer1
Specifies the message that requires the routing key to be info, error, warning.
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;
import java.io.IOException;import java.util.concurrent.TimeoutException;
public class Consumer2 {
public final static String EXCHANGE_NAME = "direct";
public static void main(String[] args) throws IOException { //创建连接、连接到RabbitMQ Connection connection = RabbitMQUtils.getConnection(); //创建一个信道 final Channel channel = connection.createChannel(); //信道设置交换器类型(direct) channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //声明一个随机队列 String queueName = channel.queueDeclare().getQueue(); //绑定 String[] routeKeys = {"info", "warning", "error"}; for (String routekey : routeKeys) { channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routekey); } System.out.println("队列[" + queueName + "]等待消息："); // 创建队列消费者 final 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("接收消息 [" + envelope.getRoutingKey() + "] " + message); } }; channel.basicConsume(queueName, true, consumer); }}Receive message:
队列[amq.gen-thfvXuQSfXHEVFRwHKZAFA]等待消息：接收消息 [warning] warning日志接收消息 [error] error日志接收消息 [info] info日志接收消息 [warning] warning日志接收消息 [error] error日志接收消息 [info] info日志Topics-topic
Compared with Direct, Topic type Exchange can route messages to different queues according to RoutingKey. It’s just that the Topic type Exchange allows the queue to use wildcards when binding the Routing key!
- #: match one or more words
- *: matches a word
user.# # 可以匹配到 user.add user.add.batchuser.* # 只能匹配到 user.add ，不能匹配到 user.add.batch
If you are going to buy pets, the types of pets are rabbit, cat, dog, the colors of pets are white, blue, grey, and the characters of pets are A, B, C. If following the routing key rules: kind.color.character, a 3*3*3=27message such as rabbit.white.A.
Producer
import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;
import java.io.IOException;import java.util.concurrent.TimeoutException;
public class TopicProducer {
public final static String EXCHANGE_NAME = "topic";
public static void main(String[] args) throws IOException, TimeoutException { //创建连接、连接到RabbitMQ Connection connection = RabbitMQUtils.getConnection(); // 创建一个信道 Channel channel = connection.createChannel(); // 指定转发 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); //宠物种类 String[] pets = {"rabbit", "cat", "dog"}; for (int i = 0; i < 3; i++) { //宠物颜色 String[] colors = {"white", "blue", "grey"}; for (int j = 0; j < 3; j++) { //宠物性格 String[] character = {"A", "B", "C"}; for (int k = 0; k < 3; k++) { // 发送的消息 String routeKey = pets[i % 3] + "." + colors[j % 3] + "." + character[k % 3]; String message = "宠物信息：" + routeKey; channel.basicPublish(EXCHANGE_NAME, routeKey, null, message.getBytes()); System.out.println(" [x] Sent " + message); } } } // 关闭连接 channel.close(); connection.close(); }}Consumer
1. If you are running a pet store, you need all pets
routingKey = #
import com.rabbitmq.client.AMQP;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.Consumer;import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.Envelope;import java.io.IOException;public class Consumer {public static void main(String[] argv) throws IOException {//创建连接、连接到RabbitMQConnection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);//声明一个随机队列String queueName = channel.queueDeclare().getQueue();//routingKey设置为 #channel.queueBind(queueName, TopicProducer.EXCHANGE_NAME, "#");System.out.println("队列[" + queueName + "]等待消息：");// 创建队列消费者final Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("接收消息 [" + envelope.getRoutingKey() + "] " + message);}};channel.basicConsume(queueName, true, consumer);}}
Receive message:
队列[amq.gen-eaK9M1vqEtY6WjivxrzqfA]等待消息：接收消息 [rabbit.white.A] 宠物信息：rabbit.white.A接收消息 [rabbit.white.B] 宠物信息：rabbit.white.B接收消息 [rabbit.white.C] 宠物信息：rabbit.white.C接收消息 [rabbit.blue.A] 宠物信息：rabbit.blue.A接收消息 [rabbit.blue.B] 宠物信息：rabbit.blue.B......//接收所有消息，省略
2. If you just want to buy a cat, but want to know the color and character of the cat first
The consumer code is the same as above, channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME,"cat.#")you can modify it
routingKey = cat.#
receive message
队列[amq.gen-Fy0aH4610sLNLrkoJKl_uA]等待消息：接收消息 [cat.white.A] 宠物信息：cat.white.A接收消息 [cat.white.B] 宠物信息：cat.white.B接收消息 [cat.white.C] 宠物信息：cat.white.C接收消息 [cat.blue.A] 宠物信息：cat.blue.A接收消息 [cat.blue.B] 宠物信息：cat.blue.B接收消息 [cat.blue.C] 宠物信息：cat.blue.C接收消息 [cat.grey.A] 宠物信息：cat.grey.A接收消息 [cat.grey.B] 宠物信息：cat.grey.B接收消息 [cat.grey.C] 宠物信息：cat.grey.C
3. If you want to buy a cat with type A personality
routingKey = cat.*.A 或 routingKey = cat.#.A
Receive message:
队列[amq.gen-xSuwMezB1VcEhcR0SfeKGA]等待消息：接收消息 [cat.white.A] 宠物信息：cat.white.A接收消息 [cat.blue.A] 宠物信息：cat.blue.A接收消息 [cat.grey.A] 宠物信息：cat.grey.A
4. If you want to buy white pets
routingKey = #.white.#
Receive message:
队列[amq.gen-1HSVv0nTfApQ_PT98lF-qQ]等待消息：接收消息 [rabbit.white.A] 宠物信息：rabbit.white.A接收消息 [rabbit.white.B] 宠物信息：rabbit.white.B接收消息 [rabbit.white.C] 宠物信息：rabbit.white.C接收消息 [cat.white.A] 宠物信息：cat.white.A接收消息 [cat.white.B] 宠物信息：cat.white.B接收消息 [cat.white.C] 宠物信息：cat.white.C接收消息 [dog.white.A] 宠物信息：dog.white.A接收消息 [dog.white.B] 宠物信息：dog.white.B接收消息 [dog.white.C] 宠物信息：dog.white.C
5. If you want to buy pets with B character
routingKey = #.B
Receive message:
队列[amq.gen-K-XtEdYjBHwcx6nAuUwhBg]等待消息：接收消息 [rabbit.white.B] 宠物信息：rabbit.white.B接收消息 [rabbit.blue.B] 宠物信息：rabbit.blue.B接收消息 [rabbit.grey.B] 宠物信息：rabbit.grey.B接收消息 [cat.white.B] 宠物信息：cat.white.B接收消息 [cat.blue.B] 宠物信息：cat.blue.B接收消息 [cat.grey.B] 宠物信息：cat.grey.B接收消息 [dog.white.B] 宠物信息：dog.white.B接收消息 [dog.blue.B] 宠物信息：dog.blue.B接收消息 [dog.grey.B] 宠物信息：dog.grey.B
6. If you want to buy a white, C-type cat
routingKey = cat.white.C
Receive message:
队列[amq.gen-LojPv9XhqR_y5SE0wqeduA]等待消息：接收消息 [cat.white.C] 宠物信息：cat.white.C