RabbitMQ 学习笔记模式

骑猪看日落 2022-11-22 10:11 135阅读 0赞

### RabbitMQ 学习笔记 模式 ###

*  连接工具类
 *  简单模式
 *   *  生产者
     *  消费者
 *  工作模式 Work queues
 *   *  生产者
     *  消费者
 *  发布订阅
 *   *  广播 Fanot
     *   *  生产者
         *  消费者
     *  定向 Direct
     *   *  生产者
         *  消费者
     *  通配符 Topic
     *   *  生产者
         *  消费者
         *  使用通配符 \*
         *  使用通配符 \#
     *  RPC
 *  参考资料

<table> 
 <thead> 
  <tr> 
   <th align="left">模式</th> 
   <th>生产者</th> 
   <th>消费者</th> 
   <th align="left">特点</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">1. 简单模式</td> 
   <td><code>1</code>个</td> 
   <td><code>1</code>个</td> 
   <td align="left">消费者平分消息，不考虑各自处理能力。 1生产、1队列、1消费</td> 
  </tr> 
  <tr> 
   <td align="left">2. 工作模式</td> 
   <td><code>1</code>个</td> 
   <td><code>n</code>个</td> 
   <td align="left">消费者争抢消息，能者多劳。和简单相比，就是存在多个消费者。1生产、1队列、n消费</td> 
  </tr> 
  <tr> 
   <td align="left">3. 发布订阅</td> 
   <td><code>1</code>个</td> 
   <td><code>n</code>个</td> 
   <td align="left">1. 引入<code>交换机</code>分发消息。<br>2. 每个队列一个消费者。（否则相当 work 模式）<br>3. 发消息前，通过绑定让<code>队列</code>和<code>交换机</code>建立关联。</td> 
  </tr> 
  <tr> 
   <td align="left">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;3.1. 广播</td> 
   <td><code>1</code>个</td> 
   <td><code>n</code>个</td> 
   <td align="left">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;1. 一条消息，所有队列都能接受到。<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;2. 广播只是通过<code>交换机</code>把消息复制成多条，发给多个<code>队列</code>。单看每个<code>队列</code>本质上 还是<code>1条消息</code>只能被<code>1个人</code>消费。<code>消费者</code>和<code>队列</code>的关系还是<code>简单模式</code>。</td> 
  </tr> 
  <tr> 
   <td align="left">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;3.2. 定向</td> 
   <td><code>1</code>个</td> 
   <td><code>n</code>个</td> 
   <td align="left">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 1. 绑定<code>队列</code>到<code>交换机</code>时加上<code>路由key</code>。发消息时带上<code>路由key</code>，<code>路由key</code>相同的<code>队列</code>就能收到消息。<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 2. 如有两个<code>消费者</code>绑定同一个<code>队列</code>，有<code>路由key</code>明确分配给谁，就不会发生争抢。</td> 
  </tr> 
  <tr> 
   <td align="left">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;3.3. 通配符</td> 
   <td><code>1</code>个</td> 
   <td><code>n</code>个</td> 
   <td align="left">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;同上，<code>消费者</code>方面支持<code>路由key</code>通配符，匹配上就能收到消息。 <code>*</code>一个单词，<code>#</code>多个单词。<code>路由key</code>由多个单词组成的用<code>.</code>分隔</td> 
  </tr> 
  <tr> 
   <td align="left"><font>———————</font></td> 
   <td></td> 
   <td></td> 
   <td align="left"></td> 
  </tr> 
 </tbody> 
</table>

# 连接工具类 #

package com.jerry.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class ConnectionUtil { 
        public static Connection getConnection() throws IOException, TimeoutException { 
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/museum");
            connectionFactory.setUsername("ty");
            connectionFactory.setPassword("9527");
            return connectionFactory.newConnection();
        }
    }

# 简单模式 #

消费者平分消息，不管考虑谁快。  
![在这里插入图片描述][20201104110843955.png]

## 生产者 ##

package com.jerry.rabbitmq.simple;
    
    import com.museum.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /** * 简单模式：生产者，发送消息 */
    public class SimpleProducer { 
        public static final String QUEUE_NAME ="simple_queue";
    
        public static void main(String[] args) throws Exception { 
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            /** * 队列声明 * @param queue 队列名称; * @param durable 是否持久化; * @param exclusive 是否独占本连接; * @param autoDelete 是否在不使用的时候自动删除队列; * @param arguments 其它参数; */
            channel.queueDeclare(QUEUE_NAME, true, false , false, null);
    
            long startTime = System.currentTimeMillis();
    
            for (int i = 0; i < 20; i++) { 
                /** * 发送消息 * @param exchange 交换机名称，不指定使用默认交换机; * @param routingKey 路由key; * @param props 其它消息属性; * @param body 消息内容; */
                channel.basicPublish("", QUEUE_NAME, null, ("大家好，我是笨笨，笨笨的笨，笨笨的笨，谢谢！" + i).getBytes());
            }
    
            System.out.println("消息已发出，耗时秒：" + (System.currentTimeMillis() - startTime) / 1000);
    
            channel.close();
            connection.close();
        }
    }

## 消费者 ##

package com.jerry.rabbitmq.simple;
    
    import com.museum.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.*;
    
    /** * 简单模式：消费者，接收并处理消息 * 平分消息，不考虑个人能力。 */
    public class SimpleConsumer { 
        public static void main(String[] args) throws Exception { 
    
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            /** * 队列声明 * @param queue 队列名称; * @param durable 是否持久化; * @param exclusive 是否独占本连接; * @param autoDelete 不用时是否自动删除; * @param arguments 其它参数; */
            channel.queueDeclare(SimpleProducer.QUEUE_NAME, true, false, false, null);
    
            // 创建消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws UnsupportedEncodingException { 
                    System.out.println(envelope.getRoutingKey());
                    System.out.println(envelope.getExchange());
                    System.out.println(envelope.getDeliveryTag());
                    System.out.println("1号消费者收到消息： " + new String(body, "utf-8"));
                }
            };
    
            /** * 监听队列 * @param 队列名称; * @param queue 队列名称 * @param autoAck 如果服务器应该考虑发送后确认的消息，则为true; 如果服务器期望显式确认，则为false; * @param callback 消费者 */
            channel.basicConsume(SimpleProducer.QUEUE_NAME, true, consumer);
        }
    }

# 工作模式 Work queues #

`每条消息`只能被`1个人`消费，所以多个`消费者`争夺，能者多劳。  
![在这里插入图片描述][2020110411092052.png]

## 生产者 ##

与简单模式一样。略。

## 消费者 ##

多个消费者代码相同。

package com.jerry.rabbitmq.work;
    
    import com.jerry.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class WorkConsumer1 { 
    
        public static void main(String[] args) throws Exception { 
    
            Connection connection = ConnectionUtil.getConnection();
            final Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(WorkProducer.QUEUE_NAME, true, false, false, null);
    
            // 设置服务端每次发送给消费者的消息数量。为0没有限制。
            channel.basicQos(1);
    
            // 创建消费者
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者1号，收到消息： " + new String(body, "utf-8"));
    
                    /** * 确认接收一条或多条消息 (告诉服务器，这条我已经消费掉了，可以继续发消息来了) * @param deliveryTag 消息id (一个递增的 long) * @param multiple 是否批量签收，所有小于等于 deliveryTag 的消息。false 表示我们一条一条来。 */
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
    
            // 监听队列
            channel.basicConsume(WorkProducer.QUEUE_NAME, true, defaultConsumer);
    
        }
    }

除了确认消息，还可以拒绝

/** * 拒绝接收到的一条或多条消息 * @param deliveryTag 消息id * @param multiple 是否指拒绝 * @param requeue true:被拒绝的消息，还可以重新请求。false: 被拒绝的消息直接丢弃 */
    channel.basicNack(envelope.getDeliveryTag(), false, true);

# 发布订阅 #

## 广播 Fanot ##

`交换机`分发消息。后面的逻辑还是`简单模式`  
![在这里插入图片描述][20201104110939699.png]

### 生产者 ###

因为引入了`交换机`，`生产者`直接将消息发给`交换机`可以和`队列`解耦了。  
`队列、交换机`：我放到`消费者`里声明即可。

package com.jerry.rabbitmq.publishsubscribe;
    
    import com.jerry.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * 发布订阅-广播模式：发送消息 */
    public class FanoutProducer { 
    
        /** 交换机名称 */
        public static final String FANOUT_EXCHANGE ="fanout_exchange";
    
        public static void main(String[] args) throws Exception { 
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            for (int i = 0; i < 10; i++) { 
                channel.basicPublish(FANOUT_EXCHANGE, "", null, String.format("%s : %d", "生产出来", i).getBytes());
            }
    
            System.out.println("产品已发出");
    
            channel.close();
            connection.close();
        }
    }

### 消费者 ###

注意：每个`消费者`绑定一个`队列`。如果一个`队列`绑多`人`，就得争抢那就变成`work模式`了。  
`FanoutConsumer1.java`

package com.jerry.rabbitmq.publishsubscribe;
    
    import com.jerry.rabbitmq.util.ConnectionUtil;
    import com.rabbitmq.client.*;
    
    import java.io.UnsupportedEncodingException;
    
    /** * 发布订阅-广播模式：消费者 */
    public class FanoutConsumer1 { 
        /** 队列名称 */
        static String QUEUE_NAME = "fanout_queue_1";
        /** 交换机名称 */
        static String EXCHANGE_NAME = FanoutProducer.FANOUT_EXCHANGE;
    
        public static void main(String[] args) throws Exception { 
    
            Connection connection = ConnectionUtil.getConnection();
            final Channel channel = connection.createChannel();
    
            /** * 队列声明 * @param queue 队列名称; * @param durable 是否持久化; * @param exclusive 是否独占本连接; * @param autoDelete 不用时是否自动删除; * @param arguments 其它参数; */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    
            /** * 交换机声明 * @param exchange 交换机名称; * @param type 换机类型; FANOUT广播，DIRECT定向，TOPIC通配 */
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT, false, false, false, null);
    
            /** * 绑定 队列+交换机 * @param queue 队列名称 * @param exchange t交换机名称 * @param routingKey 路由key（未用到留空） */
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 设置服务端每次发送给消费者的消息数量。为0没有限制。
            channel.basicQos(1);
    
            // 创建消费者
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws UnsupportedEncodingException { 
                    System.out.println("消费者1号，收到广播消息： " + new String(body, "utf-8"));
                }
            };
    
            /** * 监听队列 * @param 队列名称; * @param queue 队列名称 * @param autoAck 如果服务器应该考虑发送后确认的消息，则为true; 如果服务器期望显式确认，则为false; * @param callback 消费者 */
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
        }
    }

`FanoutConsumer2.java` 只显示区别

/** 队列名称 */
    ...
    static String QUEUE_NAME = "fanout_queue_2";
    ...
    System.out.println("消费者2号，收到广播消息： " + new String(body, "utf-8"));
    ...

## 定向 Direct ##

`路由key`作为消息标识。

1.  两个`消费者`即使绑定同一个`队列`，绑定时设置的`路由key`不同。
2.  `生产者`发消息时会指定一个具体的`路由key`。
3.  `消费者`接收消息时，`路由key`匹配的才收。  
    ![在这里插入图片描述][20201104111036574.png]

### 生产者 ###

同广播。

### 消费者 ###

与广播的区别就是`绑定`时加上了`路由key`。`消费者`只会接收和自己`路由key`相匹配的消息。

`DirectConsumer1.java`

// 绑定 队列+交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "add");

`DirectConsumer2.java`

// 绑定 队列+交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "upd");

## 通配符 Topic ##

在前一基础上，`消费端`绑定`队列`和`交换机`时`路由key1`可以带`通配符`。  
比如：发消息时key是`user.add`,`user.upd`,`user.del`，`消费端`用`user.*`都可以接收。

*  `通配符`匹配的是`单词`，多个单词用`.`分隔。
 *  每个`消费者`一个`队列`避免争抢。  
    ![在这里插入图片描述][20201104111056562.png]

### 生产者 ###

// 交换机声明
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    // 发送消息
    channel.basicPublish(EXCHANGE_NAME, "user.add", null, "新增用户".getBytes());
    channel.basicPublish(EXCHANGE_NAME, "user.upd", null, "更新用户".getBytes());
    channel.basicPublish(EXCHANGE_NAME, "user.del", null, "删除用户".getBytes());
    channel.basicPublish(EXCHANGE_NAME, "user.vip.add", null, "新增vip用户".getBytes());
    channel.basicPublish(EXCHANGE_NAME, "user.vip.upd", null, "更新vip用户".getBytes());
    channel.basicPublish(EXCHANGE_NAME, "user.vip.del", null, "删除vip用户".getBytes());

### 消费者 ###

### 使用通配符 \* ###

`TopicConsumer1.java`

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "user.*");

收到消息：

消费者1号，收到广播消息： 新增用户
    消费者1号，收到广播消息： 更新用户
    消费者1号，收到广播消息： 删除用户

### 使用通配符 \# ###

`TopicConsumer2.java`

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "user.#");

收到消息：

消费者2号，收到广播消息： 新增用户
    消费者2号，收到广播消息： 更新用户
    消费者2号，收到广播消息： 删除用户
    消费者2号，收到广播消息： 新增vip用户
    消费者2号，收到广播消息： 更新vip用户
    消费者2号，收到广播消息： 删除vip用户

## RPC ##

略。。。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p4NTIw_size_16_color_FFFFFF_t_70]

# 参考资料 #

[RabbitMQ之basicConsume、basicCancel、basicPublish等方法详解][RabbitMQ_basicConsume_basicCancel_basicPublish]  
[RabbitMQ官网文档： Interface Channel][RabbitMQ_ Interface Channel]  
[RabbitMQ官网教程][RabbitMQ]

[SpringBoot - 学习笔记 - 整合RabbitMQ][SpringBoot - _ - _RabbitMQ]

[20201104110843955.png]: /images/20221120/cf949c967c0c4f2c854a0e87b91f7b9d.png
[2020110411092052.png]: https://img-blog.csdnimg.cn/2020110411092052.png
[20201104110939699.png]: https://img-blog.csdnimg.cn/20201104110939699.png
[20201104111036574.png]: https://img-blog.csdnimg.cn/20201104111036574.png
[20201104111056562.png]: https://img-blog.csdnimg.cn/20201104111056562.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p4NTIw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20201104111103363.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p4NTIw,size_16,color_FFFFFF,t_70
[RabbitMQ_basicConsume_basicCancel_basicPublish]: https://blog.csdn.net/yaomingyang/article/details/102636657
[RabbitMQ_ Interface Channel]: https://rabbitmq.github.io/rabbitmq-java-client/api/current/com/rabbitmq/client/Channel.html
[RabbitMQ]: https://www.rabbitmq.com/getstarted.html
[SpringBoot - _ - _RabbitMQ]: https://jerryjin.blog.csdn.net/article/details/114780840