RabbitMQ几种队列工作模式详解

今天药忘吃喽~ 2022-08-31 09:56 170阅读 0赞

## RabbitMQ工作模式详解 ##

## 环境 ##

版本  
`RabbitMQ 3.8.12Erlang 23.2.6`

安装

[mac用户请点击][mac]  
windows用户自行百度

当然您有别的选择更好，总之有环境就可！

## Work Queue ##

Work Queue也叫工作模式，生产者发消息，启动多个消费者实例来消费消息，每个消费者仅消费部分信息，可达到负载均衡的  
效果。

大白话就是一个生产者对应1个或者多个消费者，通过轮训方式发送消息到消费者消费，每个消费者消费一个消息。

如如下图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]  
废话不多说，上代码。

依赖

<dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.9.0</version>
            </dependency>

先来生产者代码

package demo2;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * 工作队列模式 * 生产者发消息，启动多个消费者实例来消费消息，每个消费者仅消费部分信息，可达到负载均衡的效果。 * * 生产者采用轮训的方式给消费者派发消息 * * 每个消费者只能消费部分信息，一个消费者消费一个消息 */
    public class Produce { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
            //amqp 协议端口
            connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //声明消息队列 名称
            // 是否可持久化
            // 排他 （队列中没消息时候，系统）
            // 是否自动删除消息队列
            // 属性map集合
            channel.queueDeclare("work.queue.biz",true,false,false,null);
    
            //声明交换器
            // 交换器名称,
            // 交换器类型，
            // 是否持久化，
            // 是否自动删除的，
            // 属性map属性
            channel.exchangeDeclare("work.exchange.biz", BuiltinExchangeType.DIRECT,true,false,false,null);
    
    
            //队列和交换器绑定 并指定路由键
            channel.queueBind("work.queue.biz","work.exchange.biz","work.test");
    
            //发送消息 交换器 路由key,属性，消息 -> amqp协议会将消息发送出去
            //发送消息 发送到交换器，指定路由key
            //channel.basicPublish("work.exchange.biz","test",null,"hello world 123".getBytes());
    
    
    
            //todo 这里我们发送多条消息，模拟消费者消费
            for (int i = 0; i < 20; i++) { 
                channel.basicPublish("work.exchange.biz","work.test",null,("消息 -》 " + i ).getBytes());
    
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

这里我们定义交换器`work.exchange.biz`队列`work.queue.biz` 以及路由键`work.test`由交换机将消息发送到指定队列

定义消费者

package demo2;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.TimeoutException;
    
    
    public class LisenterComsumer1 { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    //或者下面连接方式
    // connectionFactory.setVirtualHost("/");
    // connectionFactory.setUsername("guest");
    // connectionFactory.setPassword("guest");
    //
    // //amqp 协议端口
    // connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //todo 没有就创建队列 防止启动报错
            //队列名
            //是否持久化
            //是否排他
            //是否自动删除
            //属性参数
            channel.queueDeclare("work.queue.biz",true,false,false,null);
    
            //监听
            channel.basicConsume("work.queue.biz", new DeliverCallback() {  //收到消息用 DeliverCallback
    
                public void handle(String s, Delivery delivery) throws IOException { 
                    System.out.println("推送来的消息: " + new String(delivery.getBody()));
                }
            }, new CancelCallback() {  //忽略消息 用 CancelCallback
    
                public void handle(String s) throws IOException { 
    
                }
            });
    
    // channel.close();
    // connection.close();
    
    
        }
    }

这里我们多实例启动消费者

如下图设置  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]

`注意`  
先启动消费者可能报错，因为没有队列或者交换机，需要消费者创建队列（没有就创建，有就不创建）

> channel.queueDeclare(“work.queue.biz”,true,false,false,null);

效果如下  
先启动两个消费者后，在启动生产者，两个消费者轮训均分消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]

## 发布订阅模式 ##

发布订阅模式又称为广播模式即消息广播给所有订阅该消息的消费者。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 4]

使用fanout类型交换器，routingKey忽略。每个消费者定义生成一个队列并绑定到同一个Exchange，每个消费者都可以消费到完整的消息。

Rabbitmq中默认提供了一个fanout类型的交换器,可以通过命令`amq.*` 或者图形化界面查看

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 5]

在RabbitMQ中，生产者不是将消息直接发送给消息队列，实际上生产者根本不知道一个消息被发送到哪个队列。  
生产者只需将消息发送给交换器，交换器将消息推送给消息队列，然后交换器选择将消息追加一个还是多个队列，或者丢掉等。  
所以交换器的类型就需要定义为`"fanout`

上面也说了，可以不指定交换器名，但类型必须指定，默认提供了一个"fanout类型的交换器。

好了直接上代码

生产者

package demo3;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * 消息订阅模式，即广播模式 * * 在RabbitMQ中，生产者不是将消息直接发送给消息队列，实际上生产者根本不知道一个消息被发送到哪个队列。 * 生产者将消息发送给交换器。交换器非常简单，从生产者接收消息，将消息推送给消息队列 * * * fanout 交换器很简单，从名字就可以看出来（用风扇吹出去），将所有收到的消息发送给它知道的所有的队列。 * * * * 我们无论何时连接RabbitMQ的时候，都需要一个新的，空的队列。我们可以使用随机的名字创建队列，也可以让服务器帮我们生成随机的消息队列名字。 * * 使用消费者时候，让服务器随机帮我们创建队列名(可以防止我们在任何时候连接服务器，不需要指定队列名，用完后服务器会自动删除队列) * * */
    public class Produce { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
            //amqp 协议端口
            connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //todo 声明交换器 fanout 模式生产者端只需要声明交换器，不需要声明队列，由消费者随机生成
    
            // 交换器名称,
            // 交换器类型，
            // 是否持久化，
            // 是否自动删除的，
            // 属性map属性
            channel.exchangeDeclare("fanout.exchange.biz", BuiltinExchangeType.FANOUT,true,false,false,null);
    
            //todo 声明交换器 fanout 模式生产者路由键无效，不需要指定
            //todo 这里我们发送多条消息，模拟消费者消费
            for (int i = 0; i < 20; i++) { 
                channel.basicPublish("fanout.exchange.biz","",null,("消息 -》 " + i ).getBytes());
    
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

两个消费者LisenterComsumer1,LisenterComsumer2 代码都一样，只贴一个。

package demo3;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * 广播模式 * 消费者 监听 * 我们无论何时连接RabbitMQ的时候，都需要一个新的，空的队列。我们可以使用随机的名字创建队列，也可以让服务器帮我们生成随机的消息队列名字。 * <p> * 使用消费者时候，让服务器随机帮我们创建队列名(可以防止我们在任何时候连接服务器，不需要指定队列名，用完后服务器会自动删除队列) */
    public class LisenterComsumer1 { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    // connectionFactory.setVirtualHost("/");
    // connectionFactory.setUsername("guest");
    // connectionFactory.setPassword("guest");
    //
    // //amqp 协议端口
    // connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            String queueName = channel.queueDeclare().getQueue();
            System.out.println("随机队列名: " + queueName);
    
            // 交换器名称,
            // 交换器类型，
            // 是否持久化，
            // 是否自动删除的，
            // 属性map属性
            channel.exchangeDeclare("fanout.exchange.biz", BuiltinExchangeType.FANOUT, true, false, false, null);
    
            //队列名
            //将对队列定到交换器
            // 路由键为空
    		//在创建了消息队列和 fanout 类型的交换器之后，我们需要将两者进行绑定，让交换器将消息发送给该队列。
            channel.queueBind(queueName, "fanout.exchange.biz", "");
    
            //监听
            channel.basicConsume(queueName, new DeliverCallback() {  //收到消息用 DeliverCallback
    
                public void handle(String s, Delivery delivery) throws IOException { 
                    System.out.println("推送来的消息: " + new String(delivery.getBody()));
                }
            }, new CancelCallback() {  //忽略消息 用 CancelCallback
    
                public void handle(String s) throws IOException { 
    
                }
            });
    
    // channel.close();
    // connection.close();
    
    
        }
    }

结果如下  
我们两个消费者都监听到消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 7]

这儿是推的结果，当然拉也是可以的，我们需要从队列拉取消息

GetResponse getResponse = channel.basicGet("queue.name", false);

上面使用临时队列的好处是，当我们任何时候连接服务器的时候，都能获取到一个空的新的队列。  
当不再使用时，会自动删除该队列，并且该队列不和交换器绑定。

## 路由模式 ##

使用 direct 类型的Exchange，发N条消费并使用不同的 routingKey ，消费者定义队列并将队列、 routingKey 、Exchange绑定。此时使用 direct 模式Exchagne必须要 routingKey 完全匹配的情况下消息才会转发到对应的队列中被消费。

具体实现，消费者根据路由key完全匹配才能拿到消息，生产者发送消息到交换机，不同的消息发送指定的key。

例如：info和error的日志，我们需要对其进行不同的获取。

废话不多说了，直接上代码

生产者

package demo4;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * 路由模式 * <p> * 使用 direct 类型的Exchange，发N条消费并使用不同的 routingKey ，消费者定义队列并将队列、 routingKey 、Exchange绑定。 * 此时使用 direct 模式Exchagne必须要 routingKey 完全匹配的情况下消息才会转发到对应的队列中被消费。 */
    public class Produce { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
            //amqp 协议端口
            connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //声明交换器 交换器名称, 交换器类型，是否持久化，是否自动删除的， 属性map几乎
            channel.exchangeDeclare("direct.log.exchange.biz", BuiltinExchangeType.DIRECT, false, false, false, null);
    
            //发送消息 交换器 路由key,属性，消息 -> amqp协议会将消息发送出去
    
    
            for (int i = 0; i < 100; i++) { 
    
                if (i % 2 == 0) { 
                    channel.basicPublish("direct.log.exchange.biz", "log.error",null,  ("error 日志--- > " + i ).getBytes());
    
                } else { 
                    channel.basicPublish("direct.log.exchange.biz","log.info", null,  ("info 日志 --- > " + i ).getBytes());
    
                }
    
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

两个消费者  
info日志的消费者

package demo4;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * * 路由模式 * * 消费者定义队列并将队列、 routingKey 、Exchange绑定。此时使用 direct 模式Exchagne必须要 routingKey 完全匹配的情况下消息才会转发到对应的队列中被消费。 * * 消费者绑定不同路由key获取不同的消息 * */
    public class LisenterComsumer2 { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    // connectionFactory.setVirtualHost("/");
    // connectionFactory.setUsername("guest");
    // connectionFactory.setPassword("guest");
    //
    // //amqp 协议端口
    // connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //没有就创建队列 防止启动报错
            //队列名
            //是否持久化
            //是否排他
            //是否自动删除
            //属性参数
            channel.queueDeclare("log.info.queue.biz",true,false,false,null);
    
            //声明交换器
            channel.exchangeDeclare("direct.log.exchange.biz", BuiltinExchangeType.DIRECT, false, false, false, null);
    
            //绑定队列到交换器 路由error的日志
            channel.queueBind("log.info.queue.biz","direct.log.exchange.biz","log.info");
    
    
            //监听
            channel.basicConsume("log.info.queue.biz", new DeliverCallback() {  //确认消息用 DeliverCallback
    
                public void handle(String s, Delivery delivery) throws IOException { 
                    System.out.println("推送来的消息 info 日志 " + new String(delivery.getBody()));
                }
            }, new CancelCallback() {  //忽略消息 用 CancelCallback
    
                public void handle(String s) throws IOException { 
    
                }
            });
    
    // channel.close();
    // connection.close();
    
    
        }
    }

error日志的消费者

package demo4;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * * 路由模式 * * 消费者定义队列并将队列、 routingKey 、Exchange绑定。此时使用 direct 模式Exchagne必须要 routingKey 完全匹配的情况下消息才会转发到对应的队列中被消费。 * * 消费者绑定不同路由key获取不同的消息 * */
    public class LisenterComsumer1 { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    // connectionFactory.setVirtualHost("/");
    // connectionFactory.setUsername("guest");
    // connectionFactory.setPassword("guest");
    //
    // //amqp 协议端口
    // connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //没有就创建队列 防止启动报错
            //队列名
            //是否持久化
            //是否排他
            //是否自动删除
            //属性参数
            channel.queueDeclare("log.error.queue.biz",false,false,false,null);
    
    
            //声明交换器
            channel.exchangeDeclare("direct.log.exchange.biz", BuiltinExchangeType.DIRECT, false, false, false, null);
    
    
            //绑定队列到交换器 路由error的日志
            channel.queueBind("log.error.queue.biz","direct.log.exchange.biz","log.error");
    
    
            //监听
            channel.basicConsume("log.error.queue.biz", new DeliverCallback() {  //确认消息用 DeliverCallback
    
                public void handle(String s, Delivery delivery) throws IOException { 
                    System.out.println("推送来的消息 error 日志: " + new String(delivery.getBody()));
                }
            }, new CancelCallback() {  //忽略消息 用 CancelCallback
    
                public void handle(String s) throws IOException { 
    
                }
            });
    
    // channel.close();
    // connection.close();
    
    
        }
    }

首先分别启动Info和erro的消费者，然后启动生产者,可以看到,偶数的都在erro级别的消费消费了，奇数都在info级别的消费了。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 8]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 9]

如果说，我们需要将info的收集在一个地方，error和debug的收集在一起？或者说 info,error,debug收集在一起？或者不同地方的错误日志收集在一起？等等该怎么操作？

针对这种情况，我们需要进行模糊匹配，也就是使用Topic类型的交换器即`主题模式`

## 主题模式 ##

使用 topic 类型的交换器，队列绑定到交换器、 bindingKey 时使用通配符，交换器将消息路由转发到具体队列时会根据消息 routingKey 模糊匹配，比较灵活。

bindingKey 也必须是这种形式。 topic 类型的交换器背后原理跟 direct 类型的类似：只要队列的 bindingKey 的值与消息的 routingKey 匹配，队列就可以收到该消息。有两个不同：

1.   *  (star)匹配一个单词
2.  # 匹配0到多个单词 #

![图片来自官网][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 10]

消费者可以根据不同的路由Key获取不同的消息，比较灵活。

注意：

*  如果在 topic 类型的交换器中 bindingKey 使用 \# ，则就是 fanout 类型交换器的行为。
 *  如果在 topic 类型的交换器中 bindingKey 中不使用 \* 和 \# ，则就是 direct 类型交换器的行为。

废话不多说，上代码。

我们根据不同地区的不同类型的log日志来举例。

生产者

package demo5;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    
    /** * * * 使用 topic 类型的交换器，队列绑定到交换器、 bindingKey 时使用通配符，交换器将消息路由转 * 发到具体队列时会根据消息 routingKey 模糊匹配，比较灵活。 * * 如果在 topic 类型的交换器中 bindingKey 使用 # ，则就是 fanout 类型交换器的行为。 如 3-1 3-2 * 如果在 topic 类型的交换器中 bindingKey 中不使用 * 和 # ，则就是 direct 类型交换器的行为。 * */
    public class Produce { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
            //amqp 协议端口
            connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
    
            //声明交换器 交换器名称, 交换器类型，是否持久化，是否自动删除的， 属性map几乎
            channel.exchangeDeclare("topic.log", BuiltinExchangeType.TOPIC, false, false, false, null);
    
            String aeration  = "";
            for (int i = 0; i < 100; i++) { 
    			
    //模拟地区
                if (i % 5 == 0) { 
                    aeration = "beijing";
                } else if (i % 5 > 0 & i % 5 < 2) { 
                    aeration = "jiangsu";
                } else { 
                    aeration = "shanghai";
                }
    
                System.out.println(aeration + " -> " + i % 3);
    
    //模拟log级别
                if (i % 3 == 0) { 
                //给交换机中，分别发送不同的 路由key消息
                    channel.basicPublish("topic.log", aeration + ".log.error", null, (aeration +"error 日志--- > " + i).getBytes());
    
                } else if (i % 3 == 1) { 
                    channel.basicPublish("topic.log", aeration + ".log.info", null, (aeration +"info 日志 --- > " + i).getBytes());
    
                } else if (i % 3 == 2){ 
                    channel.basicPublish("topic.log", aeration+ ".log.debug", null, (aeration +"debug 日志 --- > " + i).getBytes());
                }
    
    
            }
    
            channel.close();
            connection.close();
    
    
        }
    }

LisenterComsumer2 获取全部地区info级别log

package demo5;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.TimeoutException;
    
    
    public class LisenterComsumer2 { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    // connectionFactory.setVirtualHost("/");
    // connectionFactory.setUsername("guest");
    // connectionFactory.setPassword("guest");
    //
    // //amqp 协议端口
    // connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
            String queueName = channel.queueDeclare().getQueue();
    
            //声明交换器
            channel.exchangeDeclare("topic.log", BuiltinExchangeType.TOPIC, false, false, false, null);
    
            //绑定队列到交换器 路由error的日志
            channel.queueBind(queueName,"topic.log","#.log.info");
    
    
            //监听
            channel.basicConsume(queueName, new DeliverCallback() {  //确认消息用 DeliverCallback
    
                public void handle(String s, Delivery delivery) throws IOException { 
                    System.out.println("推送来的消息 所有地区 info 日志 " + new String(delivery.getBody()));
                }
            }, new CancelCallback() {  //忽略消息 用 CancelCallback
    
                public void handle(String s) throws IOException { 
    
                }
            });
    
    // channel.close();
    // connection.close();
    
    
        }
    }

可以看到全部info日志

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 11]

LisenterComsumer3\_1 获取全部日志

package demo5;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.URISyntaxException;
    import java.security.KeyManagementException;
    import java.security.NoSuchAlgorithmException;
    import java.util.concurrent.TimeoutException;
    
    public class LisenterComsumer3_1 { 
    
    
        public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException, URISyntaxException { 
    
    
            //连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setUri("amqp://guest:guest@localhost:5672/%2f");
    // connectionFactory.setVirtualHost("/");
    // connectionFactory.setUsername("guest");
    // connectionFactory.setPassword("guest");
    //
    // //amqp 协议端口
    // connectionFactory.setPort(5672);
    
            //简历连接
            Connection connection = connectionFactory.newConnection();
    
            //获取通道
            Channel channel = connection.createChannel();
            String queueName = channel.queueDeclare().getQueue();
    
            //声明交换器
            channel.exchangeDeclare("topic.log", BuiltinExchangeType.TOPIC, false, false, false, null);
    
            //绑定队列到交换器 路由全部日志
            channel.queueBind(queueName,"topic.log","#.log.#");
    
    
            //监听
            channel.basicConsume(queueName, new DeliverCallback() {  //确认消息用 DeliverCallback
    
                public void handle(String s, Delivery delivery) throws IOException { 
                    System.out.println("推送来的消息 3-1 全部 日志 " + new String(delivery.getBody()));
                }
            }, new CancelCallback() {  //忽略消息 用 CancelCallback
    
                public void handle(String s) throws IOException { 
    
                }
            });
    
    // channel.close();
    // connection.close();
    
    
        }
    }

可以看到 不同地区不同级别的日志都有

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 12]

当copy 3-1一份 为 3-2 topci类型多个消费者的情况下这种模式和fanout是一致的。

如下图 3-2 也全部接收到 所有地区不同日志级别。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 13]

[mac]: https://blog.csdn.net/weixin_38361347/article/details/118915290?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/aff1e562e31e4f66b246ecc63f59886a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220829/84d9d60d7ee44375a227c5a3ca6a40ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]: /images/20220829/4bf19f65617a4c9d8990deb0b47203c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]: /images/20220829/e336ee2869ff4ea29209a840ab06affa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 4]: /images/20220829/a0a0695e659943f08d29a14c71e41159.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 5]: /images/20220829/fe25ecc8febe43379c62c47b8d8f726a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 6]: /images/20220829/66826e29e4f64ad8ad8f8082973a3dd3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 7]: /images/20220829/ab6eecd81e4c4fbd808ea6cc36211233.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 8]: /images/20220829/d3d1e983bcf34f6a950ee3328b70a98a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 9]: /images/20220829/5f5159487ab44e1786a88bd8f91b0e5b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 10]: /images/20220829/e9fbbadafb32439f8330eaf9aed33f38.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 11]: /images/20220829/8057897ae1844ee68b9f47397367f1d8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 12]: /images/20220829/ecb90056bb124dd7a87c6ef5982c24ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 13]: /images/20220829/9d200e7d667e4a6ba771f4cc1a5ef76e.png