RabbitMQ入门到精通（入门篇）

た入场券 2021-09-25 14:04 707阅读 0赞

# RabbitMQ #

## 一. 初识 ##

### 1. 什么是MQ ###

> 官方解释

MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的一种数据结构。  
指把要传输的数据（消息）放在队列中，用队列机制来实现消息传递——生产者产生消息并把消息放入队列，然后由消费者去处理。  
消费者可以到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给其推送消息。

> 个人理解

mq就是消息中间件，说白了，有点类似于中介，中间人，传话者。

就拿快递智能柜举例：

1.  快递员将快递放在快递智能柜
2.  用户根据取货码，去找到存放自己快递的快递柜

在这个过程中：

快递就是**消息**

快递员就是**生产者**

投放快递就是**生产消息**

用户就是**消费者**

取出快递就是**消费消息**

快递智能柜就是**中间件**

### 2. 几大流行MQ对比 ###

![img][]

## 二. 基础 ##

### 1.概述 ###

消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能，成为异步RPC的主要手段之一。当今市面上有很多主流的消息中间件，如老牌的ActiveMQ、RabbitMQ，炙手可热的Kafka，阿里巴巴自主开发RocketMQ等。

### 2. 消息中间件的组成 ###

**2.1 Broker**

消息服务器，作为server提供消息核心服务

**2.2 Producer**

消息生产者，业务的发起方，负责生产消息传输给broker，

**2.3 Consumer**

消息消费者，业务的处理方，负责从broker获取消息并进行业务逻辑处理

**2.4 Topic**

主题，发布订阅模式下的消息统一汇集地，不同生产者向topic发送消息，由MQ服务器分发到不同的订阅者，实现消息的 广播

**2.5 Queue**

队列，PTP模式下，特定生产者向特定queue发送消息，消费者订阅特定的queue完成指定消息的接收

**2.6 Message**

消息体，根据不同通信协议定义的固定格式进行编码的数据包，来封装业务数据，实现消息的传输

### 3. 消息中间件模式分类 ###

**3.1 点对点**

PTP点对点:使用queue作为通信载体

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1YZ20w6N-1629445988070)(https://leanote.com/api/file/getImage?fileId=5ad56d7cab64411333000bb0)\]

说明：  
消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。  
消息被消费以后，queue中不再存储，所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。

**3.2 发布/订阅**

Pub/Sub发布订阅（广播）：使用topic作为通信载体

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0m4382Ei-1629445988072)(https://leanote.com/api/file/getImage?fileId=5ad56d8bab6441153c000ab3)\]

说明：  
消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。

queue实现了负载均衡，将producer生产的消息发送到消息队列中，由多个消费者消费。但一个消息只能被一个消费者接受，当没有消费者可用时，这个消息会被保存直到有一个可用的消费者。  
topic实现了发布和订阅，当你发布一个消息，所有订阅这个topic的服务都能得到这个消息，所以从1到N个订阅者都能得到一个消息的拷贝。

### 4. 消息中间件的优势 ###

**4.1 系统解耦**

交互系统之间没有直接的调用关系，只是通过消息传输，故系统侵入性不强，耦合度低。

**4.2 提高系统响应时间**

例如原来的一套逻辑，完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成；通过MQ架构设计，就可将紧急重要（需要立刻响应）的业务放到该调用方法中，响应要求不高的使用消息队列，放到MQ队列中，供消费者处理。

**4.3 为大数据处理架构提供服务**

通过消息作为整合，大数据的背景下，消息队列还与实时处理架构整合，为数据处理提供性能支持。

**4.4 Java消息服务——JMS**

Java消息服务（Java Message Service，JMS）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。  
JMS中的P2P和Pub/Sub消息模式：点对点（point to point， queue）与发布订阅（publish/subscribe，topic）最初是由JMS定义的。这两种模式主要区别或解决的问题就是发送到队列的消息能否重复消费(多订阅)。

### 5. 消息中间件应用场景 ###

**5.1 异步通信**

有些业务不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

**5.2 解耦**

降低工程间的强依赖程度，针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口，当应用发生变化时，可以独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

**5.3 冗余**

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

**5.4 扩展性**

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。便于分布式扩容。

**5.5 过载保护**

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量无法提取预知；如果以为了能处理这类瞬间峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

**5.6 可恢复性**

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

**5.7 顺序保证**

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。

**5.8 缓冲**

在任何重要的系统中，都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行，该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。

**5.9 数据流处理**

分布式系统产生的海量数据流，如：业务日志、监控数据、用户行为等，针对这些数据流进行实时或批量采集汇总，然后进行大数据分析是当前互联网的必备技术，通过消息队列完成此类数据收集是最好的选择。

### 6 消息中间件常用协议 ###

**6.1 AMQP协议**

AMQP即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同开发语言等条件的限制。  
优点：可靠、通用

**6.2 MQTT协议**

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器（比如通过Twitter让房屋联网）的通信协议。  
优点：格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统

**6.3 STOMP协议**

STOMP（Streaming Text Orientated Message Protocol）是流文本定向消息协议，是一种为MOM(Message Oriented Middleware，面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式，允许客户端与任意STOMP消息代理（Broker）进行交互。  
优点：命令模式（非topic\\queue模式）

**6.4 XMPP协议**

XMPP（可扩展消息处理现场协议，Extensible Messaging and Presence Protocol）是基于可扩展标记语言（XML）的协议，多用于即时消息（IM）以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输，这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息，即使其操作系统和浏览器不同。  
优点：通用公开、兼容性强、可扩展、安全性高，但XML编码格式占用带宽大

**6.5 其他基于TCP/IP自定义的协议**

有些特殊框架（如：redis、kafka、zeroMq等）根据自身需要未严格遵循MQ规范，而是基于TCP\\IP自行封装了一套协议，通过网络socket接口进行传输，实现了MQ的功能。

## 三. 入门 ##

### 1. 环境配置 ###

**1.1 安装erlang语言环境；**

然后配置环境变量，和jdk步骤一抹一样；

**RabbitsMQ 3.93版本**

官方下载：https://www.rabbitmq.com/install-windows.html

云盘下载：https://www.aliyundrive.com/s/3ZQr3PXpBER

**erlang 24.0版本**

官方下载：https://www.erlang.org/downloads

云盘下载：https://www.aliyundrive.com/s/tUWUznkcuBs

erlang官网：https://www.erlang.org/
    erlang与RabbitsMQ版本参考：https://www.rabbitmq.com/which-erlang.html

**1.2 安装rabbitMQ**

无脑下一步

在开始菜单中找到rabbitMQ.start ，点击运行；

**1.3 进入图形页面**

访问http://127.0.0.1:15672 默认账户：guest 密码：guest

**1.4 如果访问报错**

cmd进入rabbitmq安装目录sbin目录（D:\app_instrall\rabbitmq\rabbitmq_server-3.9.3\sbin）下分别执行这两句
    （1） rabbitmq-plugins enable rabbitmq_management //安装图形页面插件
    （2）rabbitmqctl start_app	//重启RebbitMQ

**1.5 创建用户**

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

**1.6 创建VirtualHost**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTY2MjE5_size_16_color_FFFFFF_t_70_pic_center 1]

**1.7 给用户赋予权限**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTY2MjE5_size_16_color_FFFFFF_t_70_pic_center 2]

\***VirtualHost**：如果把RabbitMQ看成一个数据库，那么VirtualHost 就是 工作空间，RabbitMQ可以拥有很多个VirtualHost。相当于数据库中的工作空间，可以有多个，根据业务的不同，连接不同的VirtualHost，达到解耦的目的；

### 2. 简单队列（点对点） ###

功能：一个生产者P发送消息到队列Q,一个消费者C接收

点对点 一个消息只能让一个消费者消费

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

**2.1 引入依赖**

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

**2.2 连接工具类**

package com.xzz.utlis;
    
    
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class ConnectionUtil { 
    
    	/** * 获取连接 * @return Connection * @throws Exception */
    	public static Connection getConnection() throws Exception { 
    		//定义连接工厂
    		ConnectionFactory factory = new ConnectionFactory();
    		//RabbitMQ ip地址
    		factory.setHost("127.0.0.1");
    		//amqp 通讯地址
    		factory.setPort(5672);
    		//设置vhost
    		factory.setVirtualHost("test");
    		factory.setUsername("xzz");
    		factory.setPassword("123@qwe");
    		//通过工厂获取连接
    		Connection connection = factory.newConnection();
    		return connection;
    	}
    
    }

**2.3 生产者代码**

private static final String QUEUE_NAME = "test_queue";
    
        	//创建队列，发送消息
    	public static void main(String[] args) throws Exception { 
    		//获取连接
    		Connection connection = ConnectionUtil.getConnection();
    		//创建通道
    		Channel channel = connection.createChannel();
    		//声明创建队列
    		channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    		//消息内容
    		String message = "Hello World!";
    		channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
    		System.out.println("发送消息："+message);
    		//关闭连接和通道
    		channel.close();
    		connection.close();
    	}

**2.4 消费者代码**

private static final String QUEUE_NAME = "test_queue";    
    //消费者消费消息
        public static void main(String[] args) throws Exception { 
            //获取连接和通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            //声明通道
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            //定义消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            //监听队列
            channel.basicConsume(QUEUE_NAME,true,consumer);
    
            while(true){ 
                //这个方法会阻塞住，直到获取到消息
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println("接收到消息："+message);
            }
        }

## 三. 方法详解 ##

### 3.1 创建队列方法 ###

**创建队列声明channel.queueDeclare方法**

DeclareOk queueDeclare() throws IOException;
    
    DeclareOk queueDeclare (String queue , boolean durable , boolean exclusive , boolean autoDelete , Map arguments) throws IOException;

**queueDeclare**参数详解

<table> 
 <thead> 
  <tr> 
   <th>String queue</th> 
   <th><strong>队列的名称</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>boolean durable</td> 
   <td><strong>设置是否持久化，为 true 则设置队列为持久化。持久化的队列会存盘，在 服务器重启的时候可以保证不丢失相关信息。</strong></td> 
  </tr> 
  <tr> 
   <td>boolean exclusive</td> 
   <td><strong>设置是否排他，为 true 则设置队列为排他的。如果一个队列被声明为排 他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除</strong></td> 
  </tr> 
  <tr> 
   <td>boolean autoDelete</td> 
   <td><strong>设置是否自动删除，为 true 则设置队列为自动删除。自动删除的前提是: 至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会 自动删除</strong></td> 
  </tr> 
  <tr> 
   <td>Map arguments</td> 
   <td><strong>设置队列的其他一些参数</strong></td> 
  </tr> 
 </tbody> 
</table>

### 3.2 发送消息方法 ###

**发送消息channel.basicPublish方法**

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
    
    void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body)
                throws IOException;
    
     void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)  
    throws IOException;

**basicPublish**参数详解

<table> 
 <thead> 
  <tr> 
   <th>exchange</th> 
   <th>交换器名称</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>routingKey</td> 
   <td>路由键</td> 
  </tr> 
  <tr> 
   <td>props</td> 
   <td>有14个成员</td> 
  </tr> 
  <tr> 
   <td>body</td> 
   <td>消息体，payload真正需要发送的消息</td> 
  </tr> 
  <tr> 
   <td>mandatory</td> 
   <td>true时，交换器无法根据自动的类型和路由键找到一个符合条件的队列，那么RabbitMq会调用Basic.Ruturn命令将消息返回给生产都，为false时，出现上述情况消息被直接丢弃</td> 
  </tr> 
  <tr> 
   <td>immediate</td> 
   <td>true，如果交换器在消息路由到队列时发现没有任何消费者，那么 这个消息将不会存和队列，当与路由匹配的所有队列都没有消费者时，会Basic.Return返回给生产者3.0去掉了immediate 参数</td> 
  </tr> 
 </tbody> 
</table>

immediate和mandatory 都是消息传递过程中，不可达目的地是，将消息返回给生产者的功能

### 3.2 监听队列方法 ###

**监听队列channel.basicConsume方法**

channel.basicConsume(String queue, boolean autoAck, Consumer callback);

**basicConsume**参数详解

<table> 
 <thead> 
  <tr> 
   <th>String queue</th> 
   <th>队列名</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>boolean autoAck</td> 
   <td>是否自动确认消息,true自动确认,false 手动调用,建立设置为false</td> 
  </tr> 
  <tr> 
   <td>Consumer callback</td> 
   <td>消费者 DefaultConsumer建立使用，重写其中的方法</td> 
  </tr> 
 </tbody> 
</table>

### 3.3 限流策略 ###

**消费端的限流策略basicQos**

void basicQos(unit prefetchSize , ushort prefetchCount, bool global )

**basicQos**参数详解

<table> 
 <thead> 
  <tr> 
   <th>unit prefetchSize</th> 
   <th>0消息大小是否限制</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ushort prefetchCount</td> 
   <td>会告诉 RabbitMQ 不要同时给一个消费者推送多于 N 个消息，即一旦有 N 个消息还没有 ack，则该 consumer 将 block 掉，直到有消息 ack</td> 
  </tr> 
  <tr> 
   <td>bool global</td> 
   <td>true、false 是否将上面设置应用于 channel，简单点说，就是上面限制是 channel 级别的还是 consumer 级别</td> 
  </tr> 
 </tbody> 
</table>

## 四. RabbitMQ特征 ##

### 4.1 交换机 ###

生产者发送消息不会向传统方式直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列，队列在将消息以推送或者拉取方式给消费者进行消费，这和我们之前学习Nginx有点类似。

交换机的作用根据具体的路由策略分发到不同的队列中，交换机有四种类型。

<table> 
 <thead> 
  <tr> 
   <th>Direct exchange（直连交换机）</th> 
   <th>根据消息携带的路由键（routing key）将消息投递给对应队列的</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Fanout exchange（扇型交换机）</td> 
   <td>将消息路由给绑定到它身上的所有队列</td> 
  </tr> 
  <tr> 
   <td>Topic exchange（主题交换机）</td> 
   <td>队列通过路由键绑定到交换机上，然后，交换机根据消息里的路由值，将消息路由给一个或多个绑定队列</td> 
  </tr> 
  <tr> 
   <td>Headers exchange（头交换机）</td> 
   <td>类似主题交换机，但是头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。</td> 
  </tr> 
 </tbody> 
</table>

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qfVFQZcD-1629445988074)(C:\\Users\\Administrator\\Desktop\\gitee\\文档\\RabbitMQ\\img\\image-20210820145734002.png)\]

**P**：消费者

**X**：交换机

\*\*红色：\*\*队列

**C**：为消费者

### 4.2 应答模式 ###

为了确保消息不会丢失，RabbitMQ支持消息应答。

消费者发送一个消息应答，告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。

**消费者挂掉却没有发送应答**

如果一个消费者挂掉却没有发送应答，RabbitMQ会理解为这个消息没有处理完全，然后交给另一个消费者去重新处理。这样，你就可以确认即使消费者偶尔挂掉也不会丢失任何消息了。

没有任何消息超时限制；只有当消费者挂掉时，RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。

消息应答是默认打开的。我们通过显示的设置autoAsk=true关闭这种机制。

现即自动应答开，一旦我们完成任务，消费者会自动发送应答。通知RabbitMQ消息已被处理，可以从内存删除。

如果消费者因宕机或链接失败等原因没有发送ACK（不同于ActiveMQ，在RabbitMQ里，消息没有过期的概念），则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。

**案例:**

//生产者端代码不变，消费者端代码这部分就是用于开启手动应答模式的。
    
    channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
    
    //注：第二个参数值为false代表关闭RabbitMQ的自动应答机制，改为手动应答。
    
    //在处理完消息时，返回应答状态，true表示为自动应答模式。 
    
    channel.basicAck(envelope.getDeliveryTag(), false);

### 4.3 公平转发 ###

目前消息转发机制是平均分配，这样就会出现一种情况：

假设现在有两个消费者，A消费者，消费完一个消息耗时1秒，B消费者消费一个消息耗时10秒。
    
    现在有一万条消息进来，A和B各5000条。
    
    A消费完耗时：1个小时多
    
    B消费完耗时：14个小时
    
    这合理么？这不合理！！！

**为了解决这样的问题，我们可以使用basicQos方法。**

传递参数为prefetchCount= 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。

换句话说，只有在消费者空闲的时候会发送下一条信息。

调度分发消息的方式，也就是告诉RabbitMQ每次只给消费者处理一条消息，也就是等待消费者处理完毕并自己对刚刚处理的消息进行确认之后，才发送下一条消息，防止消费者太过于忙碌，也防止它太过去清闲。

通过 设置channel.basicQos(1);

## 五. 发布订阅 ##

### 5.1 扇型交换机 ###

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

**这个可能是消息队列中最重要的队列了，其他的都是在它的基础上进行了扩展。**

功能实现：一个生产者发送消息，多个消费者获取消息（同样的消息），包括一个生产者，一个交换机，多个队列，多个消费者。

**思路解读（重点理解）：**

（1）一个生产者，多个消费者

（2）每一个消费者都有自己的一个队列

（3）生产者没有直接发消息到队列中，而是发送到交换机

（4）每个消费者的队列都绑定到交换机上

（5）消息通过交换机到达每个消费者的队列

该模式就是Fanout Exchange（扇型交换机）将消息路由给绑定到它身上的所有队列

以用户发邮件案例讲解

注意：交换机没有存储消息功能，如果消息发送到**没有绑定消费队列的交换机**，**消息则丢失。**

**5.2 生产者代码**

package com.xzz.level_two;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.xzz.utlis.ConnectionUtil;
    
    public class ProducerFanout { 
       private static final String EXCHANGE_NAME = "fanout_exchange";
    
       public static void main(String[] args) throws Exception{ 
               Producer(1);
       }
    
       public static void Producer(int i) throws Exception { 
           // 1.创建新的连接
           Connection connection = ConnectionUtil.getConnection();
           // 2.创建通道
           Channel channel = connection.createChannel();
           // 3.绑定的交换机 参数1交互机名称 参数2 exchange类型
           channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
           String msg = "生产者发送消息-----"+i;
           // 4.发送消息
           channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
           System.out.println("生产者发送msg：" + msg);
           // // 5.关闭通道、连接
           channel.close();
           connection.close();
           // 注意：如果消费没有绑定交换机和队列，则消息会丢失
       }
    
    }

**5.3 消费者代码**

package com.xzz.level_two;
    
    import com.rabbitmq.client.*;
    import com.xzz.utlis.ConnectionUtil;
    
    
    import java.io.IOException;
    
    public class ConsumerEmailFanout { 
       private static final String QUEUE_NAME = "consumerFanout_email";
       private static final String EXCHANGE_NAME = "fanout_exchange";
    
       public static void main(String[] args) throws Exception { 
          Consumer();
       }
    
       public static void Consumer() throws Exception { 
          System.out.println("邮件消费者启动01");
          // 1.创建新的连接
          Connection connection = ConnectionUtil.getConnection();
          // 2.创建通道
          Channel channel = connection.createChannel();
          // 3.消费者关联队列
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
          // 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey
          channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
          DefaultConsumer consumer = new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                     throws IOException { 
                String msg = new String(body, "UTF-8");
                System.out.println("01消费者获取生产者消息:" + msg);
             }
          };
          // 5.消费者监听队列消息
          channel.basicConsume(QUEUE_NAME, true, consumer);
       }
    }

### 5.2 直连交换机(路由模式) ###

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

生产者发送消息到交换机并指定一个路由key，消费者队列绑定到交换机时要制定路由key（key匹配就能接受消息，key不匹配就不能接受消息）

例如：我们可以把路由key设置为insert ，那么消费者队列key指定包含insert才可以接收消息，消费者队列key定义为update或者delete就不能接收消息。很好的控制了更新，插入和删除的操作。

**采用交换机direct模式**

**5.2.1 生产者**

public class ProducerFanout { 
       private static final String EXCHANGE_NAME = "fanout_direct";
    
       public static void main(String[] args) throws Exception{ 
          // 1.创建新的连接
          Connection connection = MQConnection.newConnection();
          // 2.创建通道
          Channel channel = connection.createChannel();
          // 3.绑定的交换机 参数1交互机名称 参数2 exchange类型
          channel.exchangeDeclare(EXCHANGE_NAME, "direct");
          String routingKey = "error";
          String msg = "fanout_exchange_msg"+routingKey;
          // 4.发送消息
          channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
           System.out.println("生产者发送msg：" + msg);
          // // 5.关闭通道、连接
           channel.close();
           connection.close();
          // 注意：如果消费没有绑定交换机和队列，则消息会丢失
       }
    }

**5.2.2 邮件消费者**

public class ConsumerEmailFanout { 
       private static final String QUEUE_NAME = "consumerFanout_email";
       private static final String EXCHANGE_NAME = "fanout_direct";
    
       public static void main(String[] args) throws Exception { 
          System.out.println("邮件消费者启动");
          // 1.创建新的连接
          Connection connection = MQConnection.newConnection();
          // 2.创建通道
          Channel channel = connection.createChannel();
          // 3.消费者关联队列
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
          // 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey
          channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
          channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
          DefaultConsumer consumer = new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                   throws IOException { 
                String msg = new String(body, "UTF-8");
                System.out.println("消费者获取生产者消息:" + msg);
             }
          };
          // 5.消费者监听队列消息
          channel.basicConsume(QUEUE_NAME, true, consumer);
       }
    }

**5.2.3 短信消费者**

public class ConsumerSMSFanout { 
       private static final String QUEUE_NAME = "ConsumerFanout_sms";
       private static final String EXCHANGE_NAME = "fanout_direct";
    
       public static void main(String[] args) throws Exception { 
          System.out.println("短信消费者启动");
          // 1.创建新的连接
          Connection connection = MQConnection.newConnection();
          // 2.创建通道
          Channel channel = connection.createChannel();
          // 3.消费者关联队列
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
          // 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey
          channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
          DefaultConsumer consumer = new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                   throws IOException { 
                String msg = new String(body, "UTF-8");
                System.out.println("消费者获取生产者消息:" + msg);
             }
          };
          // 5.消费者监听队列消息
          channel.basicConsume(QUEUE_NAME, true, consumer);
       }
    
    }

### 5.3 通配符模式 ###

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

说明：此模式是在路由key模式的基础上，使用了通配符来管理消费者接收消息。

生产者P发送消息到交换机X，type=topic，交换机根据绑定队列的routing key的值进行通配符匹配；

符号\#：匹配一个或者多个词lazy.\# 可以匹配lazy.irs或者lazy.irs.cor

符号\*：只能匹配一个词lazy.\* 可以匹配lazy.irs或者lazy.cor

## 六. 确认机制 ##

**问题产生背景:**

生产者发送消息出去之后，不知道到底有没有发送到RabbitMQ服务器， 默认是不知道的。而且有的时候我们在发送消息之后，后面的逻辑出问题了，我们不想要发送之前的消息了，需要撤回该怎么做。

**解决方案:**

1.AMQP 事务机制

2.Confirm 模式

### 6.1 事务模式 ###

txSelect 将当前channel设置为transaction模式

txCommit 提交当前事务

txRollback 事务回滚

**6.1.1 生产者**

public class Producer { 
       private static final String QUEUE_NAME = "test_queue";
    
       public static void main(String[] args) throws Exception{ 
          // 1.获取连接
          Connection newConnection = MQConnection.newConnection();
          // 2.创建通道
          Channel channel = newConnection.createChannel();
          // 3.创建队列声明
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
          // 将当前管道设置为 txSelect 将当前channel设置为transaction模式 开启事务
          channel.txSelect();
          String msg = "test_yushengjun110";
          try { 
             // 4.发送消息
             channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
             int i = 1 / 0;
             channel.txCommit();// 提交事务
             System.out.println("生产者发送消息:" + msg);
          } catch (Exception e) { 
             System.out.println("消息进行回滚操作");
             channel.txRollback();// 回滚事务
          } finally { 
             channel.close();
             newConnection.close();
          }
       }
    }

**6.1.2 消费者**

public class Customer { 
       private static final String QUEUE_NAME = "test_queue";
    
       public static void main(String[] args) throws Exception{ 
          // 1.获取连接
          Connection newConnection = MQConnection.newConnection();
          // 2.获取通道
          final Channel channel = newConnection.createChannel();
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
          DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                   throws IOException { 
                String msgString = new String(body, "UTF-8");
                System.out.println("消费者获取消息:" + msgString);
             }
          };
          // 3.监听队列
          channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
       }
    }

## 七. SpringBoot 整合RabbitMQ ##

### 7.1 maven依赖 ###

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.xzz</groupId>
        <artifactId>RabbitMQ</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            
            <java.version>1.8</java.version>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    
            
            <fastjson.version>1.2.54</fastjson.version>
            <jjwt.version>0.9.1</jjwt.version>
            <druid-starter>1.1.10</druid-starter>
            <mybatis-plus-boot-starter.version>3.4.0</mybatis-plus-boot-starter.version>
            <hutool.version>5.7.3</hutool.version>
            <aliyun-sdk-oss>3.4.0</aliyun-sdk-oss>
            <easypoi.version>4.0.0</easypoi.version>
            <lombok.version>1.18.12</lombok.version>
        </properties>
    
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.2.5.RELEASE</version>
        </parent>
    
        <dependencies>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
            </dependency>
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>${ fastjson.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>${ lombok.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.10.0</version>
            </dependency>
    
        </dependencies>
    </project>

### 7.2 配置类 ###

spring:
      rabbitmq:
      ####连接地址
        host: 127.0.0.1
       ####端口号   
        port: 5672
       ####账号 
        username: guest
       ####密码  
        password: guest
       ### 地址
        virtual-host: /

### 7.3 定义交换机和队列 ###

@Component
    public class FanoutConfig { 
    
    	// 邮件队列
    	private String FANOUT_EMAIL_QUEUE = "fanout_eamil_queue";
    
    	// 短信队列
    	private String FANOUT_SMS_QUEUE = "fanout_sms_queue";
    	// 交换机
    	private String EXCHANGE_NAME = "fanoutExchange";
    
    	// 1.定义队列邮件
    	@Bean
    	public Queue fanOutEamilQueue() { 
    		return new Queue(FANOUT_EMAIL_QUEUE);
    	}
    	// 1.定义队列短信
    	@Bean
    	public Queue fanOutSmsQueue() { 
    		return new Queue(FANOUT_SMS_QUEUE);
    	}
    
    	// 2.定义交换机
    	@Bean
    	FanoutExchange fanoutExchange() { 
    		return new FanoutExchange(EXCHANGE_NAME);
    	}
    
    	// 3.队列与交换机绑定邮件队列
    	@Bean
    	Binding bindingExchangeEamil(Queue fanOutEamilQueue, FanoutExchange fanoutExchange) { 
    		return BindingBuilder.bind(fanOutEamilQueue).to(fanoutExchange);
    	}
    
    	// 4.队列与交换机绑定短信队列
    	@Bean
    	Binding bindingExchangeSms(Queue fanOutSmsQueue, FanoutExchange fanoutExchange) { 
    		return BindingBuilder.bind(fanOutSmsQueue).to(fanoutExchange);
    	}
    }

### 7.4 生产者 ###

@Component
    public class FanoutProducer { 
    	@Autowired
    	private AmqpTemplate amqpTemplate;
    
    	public void send(String queueName) { 
    		String msg = "my_fanout_msg:" + new Date();
    		System.out.println(msg + ":" + msg);
    		amqpTemplate.convertAndSend(queueName, msg);
    	}
    }

### 7.5 消费者 ###

@Component
    @RabbitListener(queues = "fanout_eamil_queue")
    public class FanoutEamilConsumer { 
    	@RabbitHandler
    	public void process(String msg) throws Exception { 
    		System.out.println("邮件消费者获取生产者消息msg:" + msg);
    	}
    }
    
    @Component
    @RabbitListener(queues = "fanout_sms_queue")
    public class FanoutSmsConsumer { 
    	@RabbitHandler
    	public void process(String msg) { 
    		System.out.println("短信消费者获取生产者消息msg:" + msg);
    	}
    }

BindingBuilder.bind(fanOutEamilQueue).to(fanoutExchange);  
\}

// 4.队列与交换机绑定短信队列
    @Bean
    Binding bindingExchangeSms(Queue fanOutSmsQueue, FanoutExchange fanoutExchange) {
    	return BindingBuilder.bind(fanOutSmsQueue).to(fanoutExchange);
    }

### 7.4 生产者
    
    ~~~java
    @Component
    public class FanoutProducer {
    	@Autowired
    	private AmqpTemplate amqpTemplate;
    
    	public void send(String queueName) {
    		String msg = "my_fanout_msg:" + new Date();
    		System.out.println(msg + ":" + msg);
    		amqpTemplate.convertAndSend(queueName, msg);
    	}
    }

### 7.5 消费者 ###

[img]: /images/20210923/ab453dd3fc944cba96fb009a259def46.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTY2MjE5_size_16_color_FFFFFF_t_70_pic_center]: /images/20210923/98211b5cfc0e4f7f8d95fc082370c871.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTY2MjE5_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20210923/23ad181efc67476e9da7a9c2c80660fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTY2MjE5_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20210923/db11bd8704074f1992d3f004c5d02dc3.png
[dfd6fb0513cb4f92808626cee8a0053f.png_pic_center]: /images/20210923/0cbe3a37134b4d60b8d873ad91a3afaa.png
[ba3d1540da3d4219b6cda921e63996ce.png_pic_center]: /images/20210923/c76007d524db484e8a6181fca15a5a46.png
[f5cddf1902dd4d7eb84e3f5fe3510043.png_pic_center]: /images/20210923/bab3afca62b74cb6843e9e20b264faaa.png
[633bd966cb254652b17657859ac1ecfa.png_pic_center]: /images/20210923/a6480e003649474387d30a396190d2a6.png