RabbitMQ --- 六种工作模式（二）

淩亂°似流年 2022-11-26 14:57 203阅读 0赞

### 文章目录 ###

*  六种工作模式（二）
 *   *  1.4 路由模式
     *   *  1.4.1 绑定 Bindings
         *  1.4.2 直连交换机 Direct exchange
         *  1.4.3 多重绑定 Multiple bindings
         *  1.4.4 发送日志
         *  1.4.5 订阅
         *  1.4.6 完整的代码
         *   *  1、生产者
             *  2、消费者
     *  1.5 主题模式
     *   *  1.5.1 主题交换机 Topic exchange
         *  1.5.2 完整代码
         *   *  1、生产者
             *  2、消费者

# 六种工作模式（二） #

## 1.4 路由模式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 1]  
在上一小节，我们构建了一个简单的日志系统。我们能够向多个接收者广播日志消息。

在这一节，我们将向其添加一个特性—我们将只订阅所有消息中的一部分。例如，我们只接收关键错误消息并保存到日志文件(以节省磁盘空间)，同时仍然能够在控制台上打印所有日志消息。

### 1.4.1 绑定 Bindings ###

在上一节，我们已经创建了队列与交换机的绑定。使用下面这样的代码:

ch.queueBind(queueName, "logs", "");

绑定是交换机和队列之间的关系。这可以简单地理解为:队列对来自此交换的消息感兴趣。

绑定可以使用额外的routingKey参数。为了避免与basic\_publish参数混淆，我们将其称为bindingKey。这是我们如何创建一个键绑定:

ch.queueBind(queueName, EXCHANGE_NAME, "black");

bindingKey的含义取决于交换机类型。我们前面使用的fanout交换机完全忽略它。

### 1.4.2 直连交换机 Direct exchange ###

上一节中的日志系统向所有消费者广播所有消息。我们希望扩展它，允许根据消息的严重性过滤消息。例如，我们希望将日志消息写入磁盘的程序只接收关键error，而不是在warning或info日志消息上浪费磁盘空间。

前面我们使用的是fanout交换机，这并没有给我们太多的灵活性——它只能进行简单的广播。

我们将用直连交换机(Direct exchange)代替。它背后的路由算法很简单——消息传递到bindingKey与routingKey完全匹配的队列。为了说明这一点，请考虑以下设置  
![在这里插入图片描述][20200812144245536.png_pic_center]  
其中我们可以看到直连交换机X，它绑定了两个队列。第一个队列用绑定键orange绑定，第二个队列有两个绑定，一个绑定black，另一个绑定键green。

这样设置，使用路由键orange发布到交换器的消息将被路由到队列Q1。带有black或green路由键的消息将转到Q2。而所有其他消息都将被丢弃。

### 1.4.3 多重绑定 Multiple bindings ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 2]  
使用相同的bindingKey绑定多个队列是完全允许的。如图所示，可以使用binding key black将X与Q1和Q2绑定。在这种情况下，直连交换机的行为类似于fanout，并将消息广播给所有匹配的队列。一条路由键为black的消息将同时发送到Q1和Q2。

### 1.4.4 发送日志 ###

我们将在日志系统中使用这个模型。我们把消息发送到一个Direct交换机，而不是fanout。我们将提供日志级别作为routingKey。这样，接收程序将能够选择它希望接收的级别。让我们首先来看发出日志。

和前面一样，我们首先需要创建一个exchange:

//参数1: 交换机名
    //参数2: 交换机类型
    ch.exchangeDeclare("direct_logs", "direct");

接着来看发送消息的代码

//参数1: 交换机名
    //参数2: routingKey, 路由键,这里我们用日志级别,如"error","info","warning"
    //参数3: 其他配置属性
    //参数4: 发布的消息数据 
    ch.basicPublish("direct_logs", "error", null, message.getBytes());

### 1.4.5 订阅 ###

接收消息的工作原理与前面章节一样，但有一个例外——我们将为感兴趣的每个日志级别创建一个新的绑定, 示例代码如下:

ch.queueBind(queueName, "logs", "info");
    ch.queueBind(queueName, "logs", "warning");

### 1.4.6 完整的代码 ###

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

#### 1、生产者 ####

package m4_routing;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.util.Scanner;
    
    public class Producer { 
        public static void main(String[] args) throws Exception { 
            //1.连接
            ConnectionFactory f = new ConnectionFactory();
            f.setHost("192.168.126.129");//写rabbitmq服务器地址
            f.setPort(5672);//5672是通信端口，收发消息是5672，15672是管理界面
            f.setUsername("admin");
            f.setPassword("admin");
    
            //建立通道
            Channel c = f.newConnection().createChannel();
            //2.定义direct类型交换机:direct_logs
            c.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);
    
            //3.发送消息,在消息上携带路由键关键词
            while (true){ 
                System.out.println("输入消息:");
                String msg = new Scanner(System.in).nextLine();
                System.out.println("输入路由键:");
                String key = new Scanner(System.in).nextLine();
    
                /* 简单模式和路由模式，第二个参数是队1列名 c.basicPublish("", "helloworld"。。。。); 发布订阅模式，第二个参数无效 c.basicPublish("logs", ""。。。。); */
    
                // 第二个参数是路由键，通过键的匹配确定向哪个队列发送
                c.basicPublish("direct_logs",key,null,msg.getBytes());
            }
        }
    }

#### 2、消费者 ####

package m4_routing;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.Scanner;
    import java.util.UUID;
    
    public class Consumer { 
        public static void main(String[] args) throws Exception { 
            //1.连接
            ConnectionFactory f = new ConnectionFactory();
            f.setHost("192.168.126.129");//写rabbitmq服务器地址
            f.setPort(5672);//5672是通信端口，收发消息是5672，15672是管理界面
            f.setUsername("admin");
            f.setPassword("admin");
    
            //建立通道
            Channel c = f.newConnection().createChannel();
    
            /** * 三步操作： * 1.定义交换机 * 2.定义随机队列 * 3.绑定 */
            //定义交换机
            c.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT);
            //定义随机队列
            String queue = UUID.randomUUID().toString();
            c.queueDeclare(queue,false,true,true,null);
            //绑定
            System.out.println("输入绑定键，用空格隔开：");
            String s = new Scanner(System.in).nextLine();
            String[] keys = s.split("\\s+");
            for (String key:keys) { 
                c.queueBind(queue,"direct_logs",key);
            }
    
            DeliverCallback deliverCallback = new DeliverCallback() { 
    
                @Override
                public void handle(String s, Delivery message) throws IOException { 
                    String msg = new String(message.getBody());
                    String key = message.getEnvelope().getRoutingKey();
    
                    System.out.println(key+" - "+msg);
                }
            };
            CancelCallback cancelCallback = new CancelCallback() { 
    
                @Override
                public void handle(String s) throws IOException { 
    
                }
            };
    
            //3.从随机队列接受消息
            c.basicConsume(queue,true, deliverCallback, cancelCallback);
        }
    }

消费者启动多次  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 4]  
测试[http://192.168.126.129:15672][http_192.168.126.129_15672]

![在这里插入图片描述][2020081215105067.png_pic_center]

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

## 1.5 主题模式 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 6]  
在上一小节，我们改进了日志系统。我们没有使用只能进行广播的fanout交换机，而是使用Direct交换机，从而可以选择性接收日志。

虽然使用Direct交换机改进了我们的系统，但它仍然有局限性——它不能基于多个标准进行路由。

在我们的日志系统中，我们可能不仅希望根据级别订阅日志，还希望根据发出日志的源订阅日志。

这将给我们带来很大的灵活性——我们可能只想接收来自“cron”的关键错误，但也要接收来自“kern”的所有日志。

要在日志系统中实现这一点，我们需要了解更复杂的Topic交换机。

### 1.5.1 主题交换机 Topic exchange ###

发送到Topic交换机的消息,它的的routingKey,必须是由点分隔的多个单词。单词可以是任何东西，但通常是与消息相关的一些特性。几个有效的routingKey示例:“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”。routingKey可以有任意多的单词，最多255个字节。

bindingKey也必须采用相同的形式。Topic交换机的逻辑与直连交换机类似——使用特定routingKey发送的消息将被传递到所有使用匹配bindingKey绑定的队列。bindingKey有两个重要的特殊点:

*  `*` 可以通配单个单词。
 *  `#` 可以通配零个或多个单词。

用一个例子来解释这个问题是最简单的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 7]  
在本例中，我们将发送描述动物的消息。这些消息将使用由三个单词(两个点)组成的routingKey发送。routingKey中的第一个单词表示速度，第二个是颜色,第三个是物种:`“<速度>.<颜色>.<物种>”`。

我们创建三个绑定:Q1与bindingKey “`*.orange.*`” 绑定。和Q2是 “`*.*.rabbit`” 和 “`lazy.#`” 。

这些绑定可概括为:

*  Q1对所有橙色的动物感兴趣。
 *  Q2想接收关于兔子和慢速动物的所有消息。

将routingKey设置为"`quick.orange.rabbit`"的消息将被发送到两个队列。消息 "`lazy.orange.elephant`“也发送到它们两个。另外”`quick.orange.fox`“只会发到第一个队列，”`lazy.brown.fox`“只发给第二个。”`lazy.pink.rabbit`“将只被传递到第二个队列一次，即使它匹配两个绑定。”`quick.brown.fox`"不匹配任何绑定，因此将被丢弃。

如果我们违反约定，发送一个或四个单词的信息，比如"`orange`“或”`quick.orange.male.rabbit`"，会发生什么?这些消息将不匹配任何绑定，并将丢失。

另外，"`lazy.orange.male.rabbit`"，即使它有四个单词，也将匹配最后一个绑定，并将被传递到第二个队列。

### 1.5.2 完整代码 ###

#### 1、生产者 ####

package m5_topic;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.util.Scanner;
    
    public class Producer { 
        public static void main(String[] args) throws Exception{ 
            //1.连接
            ConnectionFactory f = new ConnectionFactory();
            f.setHost("192.168.126.129");//写rabbitmq服务器地址
            f.setPort(5672);//5672是通信端口，收发消息是5672，15672是管理界面
            f.setUsername("admin");
            f.setPassword("admin");
    
            //建立通道
            Channel c = f.newConnection().createChannel();
    
            //定义交换机
            c.exchangeDeclare("topic_logs", BuiltinExchangeType.TOPIC);
    
            //发送消息，携带路由键
            while (true){ 
                System.out.println("输入消息：");
                String msg = new Scanner(System.in).nextLine();
                System.out.println("输入路由键：");
                String key = new Scanner(System.in).nextLine();
                c.basicPublish("topic_logs",key,null,msg.getBytes());
            }
        }
    }

#### 2、消费者 ####

package m5_topic;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.Scanner;
    import java.util.UUID;
    
    public class Consumer { 
        public static void main(String[] args) throws Exception{ 
            //1.连接
            ConnectionFactory f = new ConnectionFactory();
            f.setHost("192.168.126.129");//写rabbitmq服务器地址
            f.setPort(5672);//5672是通信端口，收发消息是5672，15672是管理界面
            f.setUsername("admin");
            f.setPassword("admin");
    
            //建立通道
            Channel c = f.newConnection().createChannel();
    
            /** * 三步操作： * 1.定义交换机 * 2.定义随机队列 * 3.绑定 */
            //定义交换机
            c.exchangeDeclare("topic_logs", BuiltinExchangeType.TOPIC);
            //定义队列
            String queue = UUID.randomUUID().toString();
            c.queueDeclare(queue,false,true,true,null);
            //做绑定
            System.out.println("输入绑定键，空格隔开：");
            String s = new Scanner(System.in).nextLine();
            String[] keys = s.split("\\s+");
            for (String key:keys) { 
                c.queueBind(queue,"topic_logs",key);
            }
    
            DeliverCallback deliverCallback = new DeliverCallback() { 
                @Override
                public void handle(String s, Delivery message) throws IOException { 
                    String msg = new String(message.getBody());
                    String key = message.getEnvelope().getRoutingKey();
                    System.out.println(key + " - "+ msg);
                }
            };
            CancelCallback cancelCallback = new CancelCallback() { 
                @Override
                public void handle(String s) throws IOException { 
                }
            };
    
            //3.消费数据
            c.basicConsume(queue,true, deliverCallback, cancelCallback);
        }
    }

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/e7bc7fbd0cb54aa8a4209b66f41f3939.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/f6db4fa77ac94ed5948737efefc1c80c.png
[20200812144245536.png_pic_center]: /images/20221124/14efb8c03a16485097c9f7ac5e7a76f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221124/821a069ec86b4a29a98bb6810b0f593e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221124/5ebed2e8d7b140828078932c2957619c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221124/410106a40e0b414dbaffc67f9c98a6b8.png
[http_192.168.126.129_15672]: http://192.168.126.129:15672
[2020081215105067.png_pic_center]: /images/20221124/73cf2da2b48848ec9e779ce9fec9ae13.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221124/732239b639844486af02f9bbf255073b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221124/125305a896d546bebb8f73cd5fd2db4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221124/8604c1ab888f450399f6a7e676da5411.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTgwODgxNDAyNQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221124/c4ddae592b434007b197a4e8541ce9f8.png