福州网站建设优质服务商宁波网络营销策划公司

RabbitMQ 工作机制图：

Connection： 代表客户端（包括消息生产者和消费者）与RabbitMQ之间的连接。
Channel： 连接内部的Channel。channel：通道
Exchange： 充当消息交换机的组件。
Queue： 消息队列。

★ 消费消息

使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下：

（1）创建ConnectionFactory，设置连接信息，再通过ConnectionFactory获取Connection。

（2）通过Connection获取Channel。

（3）根据需要，调用Channel的queueDeclare()方法声明队列，如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列。

（4）调用Channel 的 basicConsume()方法开始处理消息，调用该方法时需要传入一个Consumer参数，
该参数相当于JMS中的消息监听器。

这个 basicConsume()方法 相当于是异步消费。
而同步消费会出现阻塞情况，这就失去消息中间件存在的意义，所以先讲异步消费。

★ 发送消息

使用RabbitMQ Java Client依赖库开发消息生产者的大致步骤如下：

（1）创建ConnectionFactory，设置连接信息，再通过ConnectionFactory获取Connection。

（2）通过Connection获取Channel。

（3）根据需要调用exchangeDeclare()、queueDeclare()方法声明Exchange和队列、并完成队列与Exchange的绑定。
如果声明的Exchange还不存在，则创建该Exchange；否则直接使用已有的Exchange。
Declare:声明、宣布

（4）调用Channel的basicPublish()方法发送消息，调用该方法的第一个参数是exchange，
第二个参数为路由key，最后两个参数依次是消息属性和消息数据体。

【注意】：虽然消息生产者与队列是完全隔离的， 但如果消息生产者不声明消息队列，那系统中就可能暂时还没有任何消息队列。

在这种情况下，消息生产者向Exchange发送的消息将不会分发给任何队列，这些消息直接就被丢弃了。

【备注】：为了保证消息生产者能将消息发送到指定队列，消息生产者需要声明消息队列，保证消息队列的存在。

**问题：**消息生产者 和 消息队列 是完全隔离的，但是生产者为什么还要声明消息队列？
**原因：**因为程序如果先运行消息生产者，后运行消费者，而声明消息队列的方法又只存在消费者那边，那么在先运行消息生产者时，就会因为还没有生成消息队列，所以生产者发送到exchange的消息，会因为没有对应的消息队列而被丢弃。

代码演示：

先创建一个普通的 maven 项目。

添加一些属性 和 RabbitMQ的依赖

创建消息消费者

把创建连接的代码封装到一个方法里去。

消费者的代码

注意：channel.basicConsume 的第二个参数 autoAck：true，就是表示自动确认消息已经被消费完成了。就是当消费者接收到消息之后，就立马返回一个已经确认消费的消息回去给消息队列。
这样容易出现问题，就是消费者这边因为一收到消息就会自动确认消息被消费了并返回已经消费消息的结果回去给消息队列，但是可能消费者其实还没有把消息消费掉，而消息队列那边又以为消费者已经把消息消费了，所以就继续发消息给那个消费者。
而消费者一收到消息又自动确认消费并返回，就会导致这个消息队列的消息越来越多，然后消费者消费不完。

这个演示已消费未确认的演示放最后

执行消费者

控制台查看
原本没有这个消息队列，通过调用Channel的queueDeclare()方法声明队列，如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列

一开始消费者声明的这个消息队列，这个是否独占的exclusive 参数我是写true，
所以下图的 myQueue01的 Features 就是 Excl

这个就是创建的消费者。
用于消费这个 myQueue01 消息队列的消费者。

后面把 exclusive 改成了 false，是因为后面的生产者，需要也声明这个 myQueue01 消息队列，而如果这个消息队列是 独占的，就没法声明了，所以改成 false

创建消息生产者

生产者发送完消息就会关闭资源
消费者则是一直启动着

测试

先启动消费者或者启动生产者都一样，因为生产者和消费者都有调用queueDeclare() 方法声明消息队列，所以不存在发送消息后没找到对应的消息队列而导致消息被丢弃的情况。

启动消费者

然后启动生产者
生产者发送消息

再看消费者，已经消费了一条消息了。
因为先启动消费者，所以生产者发送的消息马上被消费了，在控制台的队列就看不到了。

再测试：

先启动生产者
关闭消费者，然后启动生产者发送消息
可以看出消息已经生产发送到消息队列了

这一步的流程图

启动消费者消费消息

流程图：

已消费未确认

注意：channel.basicConsume 的第二个参数 autoAck：true，就是表示自动确认消息已经被消费完成了。就是当消费者接收到消息之后，就立马返回一个已经确认消费的消息回去给消息队列。
这样容易出现问题，就是消费者这边因为一收到消息就会自动确认消息被消费了并返回已经消费消息的结果回去给消息队列，但是可能消费者其实还没有把消息消费掉，而消息队列那边又以为消费者已经把消息消费了，所以就继续发消息给那个消费者。
而消费者一收到消息又自动确认消费并返回，就会导致这个消息队列的消息越来越多，然后消费者消费不完。

如图：因为 autoAck 为false ， 所以消费者消费消息后没有进行确认。这里的 unacked 条数就为1.

如果改成 autoAck 为false ，那么消费者消费消息的代码，要加上确认消息的方法。

这个就是手动确认消息。

完整代码：

ConnectionUtil 连接工具类

package cn.ljh.app.rabbitmq.util;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;//连接工具
public class ConnectionUtil
{//获取连接的方法public static Connection getConnection() throws IOException, TimeoutException{//创建连接工厂----这个ConnectionFactory源码可以看出有构造器，所以直接new一个出来ConnectionFactory connectionFactory =  new ConnectionFactory();//设置连接信息connectionFactory.setHost("localhost");connectionFactory.setPort(5672);connectionFactory.setUsername("ljh");connectionFactory.setPassword("123456");connectionFactory.setVirtualHost("/"); //连接虚拟主机//从连接工厂获取连接Connection connection = connectionFactory.newConnection();//返回连接return connection;}
}

P2PProducer 生产者

package cn.ljh.app.rabbitmq.producer;import cn.ljh.app.rabbitmq.consumer.P2PConsumer;
import cn.ljh.app.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;//消息生产者--使用默认的exchange
public class P2PProducer
{//（1）创建ConnectionFactory，设置连接信息，再通过ConnectionFactory获取Connection。//（2）通过Connection获取Channel。//（3）根据需要调用exchangeDeclare()、queueDeclare()方法声明Exchange和队列、并完成队列与Exchange的绑定。//    如果声明的Exchange还不存在，则创建该Exchange；否则直接使用已有的Exchange。//（4）调用Channel的basicPublish()方法发送消息，调用该方法的第一个参数是exchange，//    第二个参数为路由key，最后两个参数依次是消息属性和消息数据体。public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接Connection conn = ConnectionUtil.getConnection();//2、通过Connection获取Channel。Channel channel = conn.createChannel();//3、调用exchangeDeclare()方法声明Exchange、调用queueDeclare()方法声明队列，并完成队列与Exchange的绑定//此处打算直接使用默认的Exchange来分发消息，因此无需声明 Exchange，只需声明队列channel.queueDeclare(P2PConsumer.QUEUE_NAME, true, false, false, null);String message = "生产者发送的消息的内容";//4、调用Channel的basicPublish()方法发送消息channel.basicPublish(""/*默认的 Exchange 没有名字，所以用空的字符串*/,P2PConsumer.QUEUE_NAME/*使用队列名作为路由key，表明该消息将会被路由到该队列*/,null /*指定额外的消息的属性*/,message.getBytes(StandardCharsets.UTF_8)/*消息体必须是字节数组类型-->byte[]*/);//5、关闭资源//关闭通道channel.close();//关闭连接conn.close();}
}

P2PConsumer 消费者

package cn.ljh.app.rabbitmq.consumer;import cn.ljh.app.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;//消息消费者
public class P2PConsumer
{// 使用 RabbitMQ Java Client 开发 消息消费者 的大致步骤如下：//（1）创建ConnectionFactory连接工厂，设置连接信息，再通过ConnectionFactory获取Connection连接。//（2）通过Connection获取Channel。//（3）根据需要、调用Channel的queueDeclare()方法声明队列，  Declare:声明、宣布//    如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列。//（4）调用Channel 的 basicConsume()方法开始处理消息，调用该方法时需要传入一个Consumer参数，该参数相当于JMS中的消息监听器。//常量public final static String QUEUE_NAME = "myQueue01";public static void main(String[] args) throws IOException, TimeoutException{//1、创建连接工厂，设置连接信息，然后再通过连接工厂获取连接Connection conn = ConnectionUtil.getConnection();//2、通过Connection获取Channel 消息通道Channel channel = conn.createChannel();//3、调用 Channel 的 queueDeclare() 方法声明队列//如果声明的队列已存在，该方法直接获取已有的队列；如果声明的队列还不存在，该方法将会创建新的队列//参数1：声明的队列名； 参数2：消息队列是否持久化//参数3：是否只允许该消息消费者消费该队列的消息，为true，则其他消费者在这个myQueue01队列消息积堆过多的情况下，也无法帮忙消费。//参数4：是否自动删除（如果为true，在该队列没消息的情况下，会自动删除该队列） 参数5：填写额外的参数channel.queueDeclare(QUEUE_NAME, true, false, false, null);//4、调用Channel 的 basicConsume()方法开始处理消费消息channel.basicConsume(QUEUE_NAME/*消费这个名字的消费队列里面的消息*/,true/*消息的确认模式：是否自动确认*/,new DefaultConsumer(channel){//处理消息：当这个消息队列收到消息的时候，这个方法就会被触发。重写这个方法：@Overridepublic void handleDelivery(String consumerTag,Envelope envelope /*消息所在的信封,存放消息的exchange、路由key这些*/,AMQP.BasicProperties properties /*消息的那些属性*/,byte[] body /*body：消息的消息体*/) throws IOException{//把消息体中的消息拿出来String message = new String(body, "UTF-8");//printf：格式化输出函数   %s：输出字符串  %n：换行System.err.printf("P2PConsumer收到来自Exchange为【%s】、路由key为【%s】的消息，消息内容为%s%n",envelope.getExchange(),envelope.getRoutingKey(),message);}});}}

pom.xml

<?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>cn.ljh</groupId><artifactId>rabbitmqtest</artifactId><version>1.0.0</version><name>rabbitmqtest</name><!--  属性  --><properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><java.version>11</java.version></properties><!--  依赖  --><dependencies><!-- RabbitMQ 的依赖库 --><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.13.0</version></dependency></dependencies></project>