个人中心
文章发布
退出
作者热门文章
- Java锁的逻辑（结合对象头和ObjectMonitor）
- 还在用饼状图？来瞧瞧这些炫酷的百分比可视化新图形（附代码实现）⛵
- 自动注册实体类到EntityFrameworkCore上下文，并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
30
4
。
MQ(message queue)，从字面意思上看，本质是个队列，FIFO 先入先出，只不过队列中存放的内容是message 而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ 是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了 MQ 之后，消息发送上游只需要依赖 MQ，不用依赖其他服务.
关于文章全部示例代码： RabbitMQ_Study 。
①：流量消峰 如果订单系统最多能处理10000次/s的订单，这个处理能力应付正常时段下单绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的（服务处理慢不说，有可能会把响应方的服务搞宕机），但是我们能限制订单超过一万后不允许用户下单。假设使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。 ②：应用解耦 以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的请求信息被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，终端用户感受不到物流系统的故障，提升系统的可用性。 ③：异步处理 有些服务间调用是异步的，例如 A 调用 B，B 需要花费很长时间执行，但是 A 需要知道 B 什么时候可以执行完，以前一般有两种方式，A 过一段时间去调用 B 的查询 api 查询。或者 A 提供一个 callback api， B 执行完之后调用 api 通知 A 服务。这两种方式都不是很优雅，使用消息总线，可以很方便解决这个问题，A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样 B 服务也不用做这些操作。A 服务还能及时的得到异步处理成功的消息.
RabbitMQ 是一个在AMQP基础上实现的，可复用的企业消息系统。它可以用于大型软件系统各个模块之间的高效通信，支持高并发，支持可扩展。 你可以把它当做一个快递站点，当你要发送一个包裹时，你把你的包裹放到快递站，快递员最终会把你的快递送到收件人那里，按照这种逻辑 RabbitMQ 是一个快递站，一个快递员帮你传递快件。RabbitMQ 与快递站的主要区别在于，它不处理快件而是 接收 ， 存储 和 转发消息数据 .
AMQP： 即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制.
消息队列： MQ 全称为Message Queue, 消息队列。是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信。队列的使用除去了接收和发送应用程序同时执行的要求。在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量.
①：生产者（Producer）
产生数据发送消息的程序是生产者
②：交换机（Exchange）
交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何
处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
③：队列（Queue）
队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中。队列仅受主机的内存和
磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是
我们使用队列的方式
④：消费者（Consumer）
消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。
同一个应用程序既可以是生产者又是可以是消费者。
Broker：
接收和分发消息的应用，RabbitMQ Server就是Message Broker；简单来说就是消息队列服务器实体Virtual host：出于多租户和
安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个
RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue等 Connection： publisher／consumer和broker之间的TCP连接 Channel： 消息通道，如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCPConnection的开销将是巨大的，效率也较
低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP
method包含了channel id帮助客户端和message broker识别channel，所以channel之间是完全隔离的。Channel作为轻量级的
Connection极大减少了操作系统建立TCP connection的开销。 Exchange： message到达broker的第一站，根据分发规则，匹配查询表中的routing key，分发消息到queue中去。
常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout(multicast) Queue： 消息队列载体，每个消息都会被投入到一个或多个队列;消息最终被送到这里等待consumer取走 Routing Key： 路由关键字，exchange根据这个关键字进行消息投递。 Binding： exchange和queue之间的虚拟连接，binding中可以包含routing key，Binding信息被保存到exchange中的查询表中，用于message
的分发依据;它的作用就是把exchange和queue按照路由规则绑定起来。 producer： 消息生产者，就是投递消息的程序。 consumer： 消息消费者，就是接受消息的程序。
这里我以RabbitMQ 3.9 的版本来进行本文的讲解，这里我们需要准备几个文件 。
RabbitMQ 3.9.13 Erlang 23.3.4.11 ( 版本兼容 ) rabbitmq_delayed_message_exchange-3.8.0.ez 。
#
安装 erlang 环境
rpm -ivh erlang-23.3.4.11-1.el7.x86_64.rpm
#
安装 socat 环境
yum -y install socat
#
安装 RabbitMQ 服务
rpm -ivh rabbitmq-server-3.9.13-1.el7.noarch.rpm
#
检查是否安装
yum list | grep rabbitmq yum list | grep erlang
注： socat支持多协议，用于协议处理，端口转发，rabbitmq依赖于socat，因此在安装rabbitmq前要安装socat。由于默认的CentOS-Base.repo源中没有socat，所以 $ yum install socat会出现以下错误： No package socat available 。
补充命令：
#
查看所以的已开启的端口
firewall-cmd --zone=public --list-ports
#
开启15672端口（--permanent代表永久生效，重启系统不会失效）
firewall-cmd --zone=public --add-port=15672/tcp --permanent
#
防火墙关闭
systemctl stop firewalld
#
启动RabbitMQ服务
systemctl start rabbitmq-server 或 /sbin/service rabbitmq-server start 或 service rabbitmq-server start
#
添加开机启动RabbitMQ服务
chkconfig rabbitmq-
server on
#
开启web管理接口（可以更方便快速的对RabbitMQ进行操作）
rabbitmq-plugins enable rabbitmq_management
#
停止RabbitMQ服务
systemctl stop rabbitmq-server 或 /sbin/service rabbitmq-server stop 或 service rabbitmq-server stop
注：web管理接口应用的操作
rabbitmqctl stop_app 停止web页面
rabbitmqctl start_app 启动web页面
若大家用的是最新版本安装可能不太一样，请前往： RabbitMQ安装 。
关于RabbitMQ中的用户角色【tags】
其他（none）：
不能登录管理控制台（启用management plugin的情况下，以下相同）
普通管理者（management）：
用户可以通过AMQP做的任何事外加以下权限
列出自己可以通过AMQP登入的virtual hosts
查看自己的virtual hosts中的queues
,
exchanges 和 bindings
查看和关闭自己的channels 和 connections
查看有关自己的virtual hosts的“全局”的统计信息，包含其他用户在这些virtual hosts中的活动
决策制定者（policymaker）：
management的权限外加以下权限
查看、创建和删除自己的virtual hosts所属的policies和parameters
监控者（monitor
/
monitoring）：
management的权限外加以下权限
列出所有virtual hosts，包括他们不能登录的virtual hosts
查看其他用户的connections和channels
查看节点级别的数据如clustering和memory使用情况
查看真正的关于所有virtual hosts的全局的统计信息
超级管理员（administrator）：
policymaker和monitoring的权限外加以下权限
创建和删除virtual hosts
查看、创建和删除users
查看创建和删除permissions
关闭其他用户的connections
#
查看RabbitMQ里的所有用户
rabbitmqctl list_users
#
查看默认guest用户的权限
rabbitmqctl list_user_permissions {username}
【RabbitMQ中的用户管理】
rabbitmqctl add_user {username} {password}
#
该命令将创建一个 non-administrative 用户
rabbitmqctl delete_user {username}
#
表示删除一个用户，该命令将指示RabbitMQ broker去删除指定的用户
rabbitmqctl change_password {username} {newpassword}
#
表示修改指定的用户的密码
rabbitmqctl clear_password {username}
#
表示清除指定用户的密码
#
执行此操作后的用户，将不能用密码登录，但是可能通过已经配置的SASL EXTERNAL的方式登录。
rabbitmqctl authenticate_user {username} {password}
#
表示指引RabbitMQ broker认证该用户和密码
rabbitmqctl set_user_tags {username} {tag ...
}
#
表示设置用户的角色，｛tag｝可以是零个，一个，或者是多个。并且已经存在的tag也将会被移除
【RabbitMQ中的权限控制】
在上面我们添加完相关的用户后，就可以对其用户分配相关vhost的权限了。
vhost对于Rabbit就像虚拟机之于物理服务器一样，它们通过在各个实例间提供逻辑上分离，
允许你为不同的应用程序安全保密地运行数据。
而在RabbitMQ中相应的权限分为读、写、配置三部分：
读：有关消费消息的任何操作，包括“清除”整个队列（同样需要绑定操作的成功）
写：发布消息（同样需要绑定操作的成功）
配置：队列和交换机的创建和删除
知道了RabbitMQ权限相关的配置后，我们就可以根据具体情况来配置相应的信息。
RabbitMQ的权限是以vhost为分隔的，我们需要确定一个vhost来确定相关的权限设置，默认的vhost是“
/
”
rabbitmqctl add_vhost {vhost}
#
｛vhost｝ 表示待创建的虚拟主机项的名称
rabbitmqctl delete_vhost {vhost}
#
表示删除一个vhost。删除一个vhost将会删除该vhost的所有exchange、queue、binding、用户权限、参数和策略。
rabbitmqctl list_vhosts {vhostinfoitem ...
}
#
表示列出所有的vhost。其中 {vhostinfoitem} 表示要展示的vhost的字段信息，展示的结果将按照{vhostinfoitem}指定的字段
#
顺序展示。这些字段包括： name（名称） 和 tracing （是否为此vhost启动跟踪）。
#
如果没有指定具体的字段项，那么将展示vhost的名称。
rabbitmqctl set_permissions [-
p vhost] {user} {conf} {write} {read}
#
表示设置用户权限。 {vhost} 表示待授权用户访问的vhost名称，默认为 "/"； {user} 表示待授权反问特定vhost的用户名称；
#
{conf}表示待授权用户的配置权限，是一个匹配资源名称的正则表达式； {write} 表示待授权用户的写权限，是一个匹配资源名称
#
的正则表达式； {read}表示待授权用户的读权限，是一个资源名称的正则表达式。
#
rabbitmqctl set_permissions -p / admin "^mip-.*" ".*" ".*"
#
例如上面例子，表示授权给用户 "admin" 具有所有资源名称以 "mip-" 开头的 配置权限；所有资源的写权限和读权限。
rabbitmqctl clear_permissions [-
p vhost] {username}
#
表示设置用户拒绝访问指定指定的vhost，vhost默认值为 "/"
rabbitmqctl list_permissions [-
p vhost]
#
表示列出具有权限访问指定vhost的所有用户、对vhost中的资源具有的操作权限。默认vhost为 "/"。
#
注意，空字符串表示没有任何权限。
实际操作说明： rabbitmqctl list_users
#
查看RabbitMQ里的所有用户
rabbitmqctl list_vhosts
#
查看RabbitMQ里的所有vhosts
rabbitmqctl list_permissions
#
查看RabbitMQ里所有用户的权限
rabbitmqctl list_user_permissions guest
#
查看RabbitMQ里guest用户的权限
rabbitmqctl add_vhost test
#
创建的一个虚拟主机项为 test 的名称
rabbitmqctl add_user admin 123
# 创建一个用户为admin 密码为123 rabbitmqctl set_user_tags admin administrator
#
设置admin的角色为超级管理员（administrator）
rabbitmqctl set_permissions -p test admin ".*" ".*" ".*"
#
设置admin在test的vhost中，并设置全部文件的读写操作
rabbitmqctl list_permissions -p test
#
查看test中的vhost里的用户
systemctl stop rabbitmq-server
#
停止RabbitMQ服务
yum list | grep rabbitmq
#
查看RabbitMQ安装的相关列表
yum -y remove rabbitmq-server.noarch
#
卸载RabbitMQ已安装的相关内容
yum list | grep erlang
#
查看erlang安装的相关列表
yum -y remove erlang-* yum remove erlang.x86_64
#
卸载erlang已安装的相关内容
rm -rf /usr/lib64/erlang rm -rf /var/lib/rabbitmq rm -rf /usr/local/erlang rm -rf /usr/local/rabbitmq
#
删除有关的所有文件
本小节将使用Java编写两个程序来模拟简单队列，用生产者（Producer）发送消息到RabbitMQ队列后，再由消费者（Consumer）来监控RabbitMQ发送来的队列信息； 简单队列就是一个生产者发送消息到队列，监听那个队列的一个消费者获取消息并处理 。
<
dependencies
>
<!--
RabbitMQ客户端坐标
-->
<
dependency
>
<
groupId
>
com.rabbitmq
</
groupId
>
<
artifactId
>
amqp-client
</
artifactId
>
<
version
>
5.14.2
</
version
>
</
dependency
>
</
dependencies
>
<
build
>
<
plugins
>
<!--
配置maven编译版本
-->
<
plugin
>
<
groupId
>
org.apache.maven.plugins
</
groupId
>
<
artifactId
>
maven-compiler-plugin
</
artifactId
>
<
version
>
3.8.1
</
version
>
<
configuration
>
<
source
>
1.8
</
source
>
<!--
源代码使用的JDK
-->
<
target
>
1.8
</
target
>
<!--
target需要生成的目标class文件的编译版本
-->
<
encoding
>
UTF-8
</
encoding
>
<!--
字符集编码，防止中文乱码
-->
<
failOnError
>
true
</
failOnError
>
<!--
指示即使存在编译错误，构建是否仍将继续
-->
<
failOnWarning
>
false
</
failOnWarning
>
<!--
指示即使存在编译警告，构建是否仍将继续
-->
<
showDeprecation
>
false
</
showDeprecation
>
<!--
设置是否显示使用不推荐API的源位置
-->
<
showWarnings
>
false
</
showWarnings
>
<!--
设为true若要显示编译警告，请执行以下操作
-->
<
meminitial
>
128M
</
meminitial
>
<!--
编译器使用的初始化内存
-->
<
maxmem
>
512M
</
maxmem
>
<!--
编译器使用的最大内存
-->
</
configuration
>
</
plugin
>
</
plugins
>
</
build
>
package
cn.xw.helloWorld;
import
com.rabbitmq.client.Channel;
import
com.rabbitmq.client.Connection;
import
com.rabbitmq.client.ConnectionFactory;
import
java.io.IOException;
import
java.nio.charset.StandardCharsets;
import
java.util.concurrent.TimeoutException;
/**
*
@author
AnHui OuYang
*
@version
1.0
* created at 2022-03-04 17:42
*/
public
class
Producer {
//
简单队列名称
public
static
final
String QUEUE_NAME = "helloWorldQueue"
;
public
static
void
main(String[] args)
throws
IOException, TimeoutException {
//
创建一个连接工厂
ConnectionFactory factory =
new
ConnectionFactory();
//
设置RabbitMQ服务的IP、账号、密码、Vhost虚拟主机(默认 "/" 则不需要设置)
factory.setHost("192.168.31.51"
);
factory.setUsername(
"admin"
);
factory.setPassword(
"123"
);
factory.setVirtualHost(
"test"
);
//
通过工厂对象获取一个连接
Connection connection =
factory.newConnection();
//
通过连接来获取一个信道
Channel channel =
connection.createChannel();
//
声明一个队列
//
参数一：队列名称
//
参数二：队列里的消息是否持久化，默认消息保存在内存中，默认false
//
参数三：该队列是否只供一个消费者进行消费的独占队列，则为 true（仅限于此连接），false（默认，可以多个消费者消费）
//
参数四：是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除，默认false
//
参数五：构建队列的其它属性，
看下面扩展参数
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
//
发送的消息
byte
[] msg = "这是一个简单消息"
.getBytes(StandardCharsets.UTF_8);
//
发送消息
//
参数一：将发送到RabbitMQ的哪个交换机上
//
参数二：路由的key是什么（直接交换机找到路由后，通过路由key来确定最终的队列）
//
参数三：其它参数
//
参数四：发送到队列的具体信息
channel.basicPublish("", QUEUE_NAME,
null
, msg);
System.out.println(
"消息发送完成！"
);
}
}
package
cn.xw.helloWorld;
import
com.rabbitmq.client.*
;
import
java.io.IOException;
import
java.nio.charset.StandardCharsets;
import
java.util.concurrent.TimeoutException;
/**
*
@author
AnHui OuYang
*
@version
1.0
* created at 2022-03-05 15:12
*/
public
class
Consumer {
//
简单队列名称
public
static
final
String QUEUE_NAME = "helloWorldQueue"
;
public
static
void
main(String[] args)
throws
IOException, TimeoutException {
//
创建一个连接工厂
ConnectionFactory factory =
new
ConnectionFactory();
//
设置RabbitMQ服务的IP、账号、密码、Vhost虚拟主机(默认 "/" 则不需要设置)
factory.setHost("192.168.31.51"
);
factory.setUsername(
"admin"
);
factory.setPassword(
"123"
);
factory.setVirtualHost(
"test"
);
//
通过工厂对象获取一个连接
Connection connection =
factory.newConnection();
//
通过连接来获取一个信道
Channel channel =
connection.createChannel();
System.out.println(
"消费者开始监听队列消息...."
);
//
推送的消息如何进行消费的接口回调
DeliverCallback deliverCallback =
new
DeliverCallback() {
@Override
public
void
handle(String consumerTag, Delivery message)
throws
IOException {
System.out.println(
"获取队列信息：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}
};
//
取消消费的一个回调接口 如在消费的时候队列被删除掉了
CancelCallback cancelCallback =
new
CancelCallback() {
@Override
public
void
handle(String consumerTag)
throws
IOException {
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
}
};
//
消费者消费消息
//
参数一：消费哪个队列
//
参数二：消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
//
参数三：接受队列消息的回调接口
//
参数四：取消消费的回调接口
channel.basicConsume(QUEUE_NAME,
true
, deliverCallback, cancelCallback);
}
}
编写好上面的消费者代码和生产者代码后我们就可以进行Demo演示了，首先执行生产者发送消息后我们再执行消费者代码 。
随后执行完消费者后会打印具体的队列消息 。
注： 必须先执行生产者，因为执行消费者后会发现在RabbitMQ中找不到指定Queue队列，这时就会出现异常；但是为了不报错也可以在消费者代码里面也创建队列，所有，生产者消费者都可以创建队列 。
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue 'helloWorldQueue' in vhost 'test', class-id=60, method-id=20) 。
x-dead-letter-exchange：
死信交换器
x-dead-letter-routing-key：
死信消息的可选路由键
x-expires：
队列在指定毫秒数后被删除
x-message-ttl：
毫秒为单位的消息过期时间，队列级别
x-ha-policy：
创建HA队列，此参数已失效
x-ha-nodes：
HA队列的分布节点，此参数已失效
x-max-length：
队列的消息条数限制。限制加入queue中消息的条数。先进先出原则，超过后，后面的消息会顶替前面的消息。
x-max-length-bytes：
消息容量限制，该参数和x-max-length目的一样限制队列的容量，但是这个是靠队列大小（bytes）来达到限制。
x-max-priority：
最大优先值为255的队列优先排序功能
x-overflow：
设置队列溢出行为。这决定了当达到队列的最大长度时消息会发生什么。
有效值是drop-head、reject-publish或reject-publish-dlx。
x-single-active-consumer：
表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，
其他被忽略，false时消息循环分发给所有消费者(默认false)
x-queue-mode：
将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用；
如果未设置，队列将保留内存缓存以尽可能快地传递消息
x-queue-master-locator：
在集群模式下设置镜像队列的主节点信息
工作队列（又称任务队列）的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务.
生产者生产了1万个消息发送到队列中，这时为了提高处理效率往往设置了多个消费者同时监听消息队列并处理消息 。
/**
*
@author
AnHui OuYang
*
@version
1.0
* created at 2022-03-05 16:24
*/
public
class
ChannelUtil {
public
static
Channel getChannel() {
//
信道初始化
Channel channel =
null
;
try
{
//
创建一个连接工厂
ConnectionFactory factory =
new
ConnectionFactory();
//
设置RabbitMQ服务的IP、账号、密码、Vhost虚拟主机(默认 "/" 则不需要设置)
factory.setHost("192.168.31.51"
);
factory.setUsername(
"admin"
);
factory.setPassword(
"123"
);
factory.setVirtualHost(
"test"
);
//
通过工厂对象获取一个连接
Connection connection =
factory.newConnection();
//
通过连接来获取一个信道
channel =
connection.createChannel();
}
catch
(Exception e) {
e.printStackTrace();
}
return
channel;
}
}
public
class
ProducerA {
//
工作队列名称
public
static
final
String QUEUE_NAME = "workQueue"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建队列
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
//
循环发送消息
for
(
int
i = 0; i < 1000; i++
) {
byte
[] msg = ("这是一个编号为：" + i + " 的待处理的消息"
).getBytes(StandardCharsets.UTF_8);
channel.basicPublish(
"", QUEUE_NAME,
null
, msg);
}
System.out.println(
"消息发送完成！"
);
}
}
public
class
ConsumerA {
//
工作队列名称
public
static
final
String QUEUE_NAME = "workQueue"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建队列 以防启动消费者发现队列不存在报错
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
System.out.println(
"消费者A开始监听队列消息...."
);
//
消费者消费消息
channel.basicConsume(QUEUE_NAME,
true
, (consumerTag, message) ->
{
System.out.println(
"A消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
ConsumerB {
//
工作队列名称
public
static
final
String QUEUE_NAME = "workQueue"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建队列 以防启动消费者发现队列不存在报错
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
System.out.println(
"消费者B开始监听队列消息...."
);
//
消费者消费消息
channel.basicConsume(QUEUE_NAME,
true
, (consumerTag, message) ->
{
System.out.println(
"B消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
创建完生产者和消费者后首先启动两个消费者，然后启动生产者，生产者发送消息，被两个消费者监听并消费 。
在上面的代码中我们会发现两个消费者消费消息的顺序是轮询的（A1，B2，A3，B4......）；这也是默认的消费规则，但是在日常生产环境中并不会用此模式来进行队列消息的消费.
每个消费者服务完成一个任务可能需要的时间长短不一样，如果其中一个消费者处理一个任务时并仅只完成了部分就突然挂掉了，会发生什么情况。RabbitMQ一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费的消息，因为它无法接收到。为了保证消息在发送过程中不丢失，rabbitmq引入消息应答机制， 消息应答就是：消费者在接收到消息并且处理该消息之后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了 。
消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡，因为这种模式如果消息在消费者接收到之前，消费者那边出现连接或者channel关闭，那么消息就丢失了,当然另一方面这种模式在消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用（就是业务处理简单的消息）.
自动应答： 队列向消费者发送消息后，消费者接收到消息就算成功应答了，随后队列将会删除对应的队列消息； 。
上面案例全部采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答，这样确保从消息队列来一个消息给消费者，等消费者消费完毕以后再告知RabbitMQ已处理完，然后RabbitMQ才会发送下一条消息个消费者处理，保证消息不丢失 。
注：basicConsume消息接收方法中的autoAck参数必须为false才可以显示为手动确认
手动应答分为三种情况:
②：手动拒绝 basicReject(long deliveryTag, boolean requeue)：
拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。
③：手动不确认 basicNack(long deliveryTag, boolean multiple, boolean requeue)
不确认deliveryTag对应的消息，第二个参数是否应用于多消息，第三个参数是否requeue，与basic.reject区别就是同时支持多个消息，
可以nack该消费者先前接收未ack的所有消息。nack后的消息也会被自己消费到。
③：手动恢复 basicRecover(boolean requeue)
是否恢复消息到队列，true则重新入队列，并且尽可能的将之前recover的消息投递给其他消费者消费，而不是自己再次消费。
false则消息会重新被投递给自己。
④：手动应答 basicAck(long deliveryTag, boolean multiple)
如果消费者在处理消息的过程中，出了错，就没有什么办法重新处理这条消息，所以在平时都是处理消息成功后，再确认消息；
当autoAck=false时，RabbitMQ会等待消费者手动发回ack信号后,才从内存（和磁盘，如果是持久化消息的话）中移除消息。
它采用消息确认机制，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，
因为RabbitMQ会一直持有消息直到消费者手动调用channel.basicAck为止。
对于RabbitMQ服务器端而言，如果服务器端一直没 有收到消费者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个 消费者（也可能还是原来的那个消费者）。
这里我们启动了手动确认后，就必须调用channel.basicAck方法进行确认，
否则的话RabbitMQ会一直进行等待，当我们这个消费者关闭后，RabbitMQ会将该条消息再发给对应的消费者进行消费，
直到有消费者对该条消息进行消费并应答完成。
参数说明： deliveryTag：
对应消息的ID；通过message.getEnvelope().getDeliveryTag()获取
requeue：
是否重新入列，true代表拒绝应答后会重新返回队列，false则直接删除或者进入死信队列
multiple：
是否批量应答，true代表批量应答
假设有个队列依次排列为 1、2、3...10 (1最先出队，10最后出队)；
当为true，发送1~5消息给消费者处理完都未确认，当到第6时执行应答方法，并且multiple为true，则代表1~6都被被批量应答
当为false，发送1~5消息给消费者处理完都未确认，当到第6时执行应答方法，并且multiple为true，则代表只要6被应答
//生产者只管发任务消息，代码不变，消费者代码优化更改以下，多个消费者代码也和这一样
public
class
ConsumerB {
//
工作队列名称
public
static
final
String QUEUE_NAME = "workQueue"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建队列 以防启动消费者发现队列不存在报错
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
System.out.println(
"消费者B开始监听队列消息...."
);
//
应答方式 true自动应答 false手动应答(
若是手动应答必须设置false
)
boolean
autoAck =
false
;
//
消费者消费消息requeue
channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) ->
{
try
{
//
这里我就一句打印语句，没有复杂逻辑，正常这里有复杂业务
System.out.println("B消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
int
i = 1/0
;
//
手动确认应答 不批量应答
channel.basicAck(message.getEnvelope().getDeliveryTag(),
false
);
}
catch
(IOException e) {
e.printStackTrace();
//
出现异常手动进行不应答;并且放入队列中（reject或者使用uack方式都可以，或者本次消息不处理了可以通过recover重新放到队列）
channel.basicReject(message.getEnvelope().getDeliveryTag(),
true
);
}
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
注："在手动应答的情况下,如果channel.basicAck收到确认前的代码有问题，会抛出异常,导致无法进行手动确认，一般消费者也不会连接中断，那么该消息就一直无法被处理，连被其它消费者处理的机会都没有，所以一般我们会进行try-catch处理，处理成功则手动确认，失败或有异常则拒绝。" 。
在生产过程中，难免会发生服务器宕机的事情，RabbitMQ也不例外，可能由于某种特殊情况下的异常而导致RabbitMQ宕机从而重启，那么这个时候对于消息队列里的数据，包括交换机、队列以及队列中存在消息恢复就显得尤为重要了。RabbitMQ本身带有持久化机制，包括交换机、队列以及消息的持久化。持久化的主要机制就是将信息写入磁盘，当RabbitMQ服务宕机重启后，从磁盘中读取存入的持久化信息，恢复数据.
默认不是持久化的，在服务器重启之后，交换机会消失。我们在管理台的Exchange页签下查看交换机，可以看到使用上述方法声明的交换机，Features一列是空的，即没有任何附加属性.
我们可以看到第三个参数durable，如果为true时则表示要做持久化，当服务重启时，交换机依然存在，所以使用该方法声明的交换机是下面这个样子的:
与交换机的持久化相同，队列的持久化也是通过durable参数实现的(设置后队列也会有个D)，看一下方法的定义：
queueDeclare(String queue,boolean durable,boolean exclusive,boolean autoDelete,Map<String, Object> arguments) boolean durable：
参数跟交换机方法的参数一样，true表示做持久化，当RabbitMQ服务重启时，队列依然存在
boolean exclusive（补充）：
排它队列。如果一个队列被声明为排他队列，那么这个队列只能被第一次声明它的连接所见，并在连接断开的时候自动删除。
这里有三点需要说明：
1：排它队列是基于连接可见的，同一连接的不同信道是可以同时访问同一连接创建的排它队列
2：如果一个连接已经声明了一个排它队列，其它连接是不允许建立同名的排它队列的，这个与普通队列不同
3：即使该队列是持久化的，一旦连接关闭或者客户端退出，该排它队列都会被自动删除的，这种队列适用于一
个客户端发送读取消息的应用场景
boolean autoDelete（补充）：
自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列
消息的持久化是指当消息从交换机发送到队列之后，被消费者消费之前，服务器突然宕机重启，消息仍然存在。消息持久化的前提是队列持久化，假如队列不是持久化，那么消息的持久化毫无意义。通过如下代码设置消息的持久化:
basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
BasicProperties props设置消息持久化方式：
参数实现类：
public
static
class
BasicProperties
extends
com.rabbitmq.client.impl.AMQBasicProperties {
private
String contentType;
//
消息的内容类型，如：text/plain
private
String contentEncoding;
//
消息内容编码
private
Map<String,Object> headers;
//
设置消息的header,类型为Map<String,Object>
private
Integer deliveryMode;
//
1（nopersistent）非持久化，2（persistent）持久化
private
Integer priority;
//
消息的优先级
private
String correlationId;
//
关联ID
private
String replyTo;
//
用于指定回复的队列的名称
private
String expiration;
//
消息的失效时间
private
String messageId;
//
消息ID
private
Date timestamp;
//
消息的时间戳
private
String type;
//
类型
private
String userId;
//
用户ID
private
String appId;
//
应用程序ID
private
String clusterId;
//
集群ID
}
deliveryMode是设置消息持久化的参数，等于1不设置持久化，等于2设置持久化；
我们平时不会使用BasicProperties类而是使用MessageProperties，通过这个类来获取具体配置
设置
MessageProperties.PERSISTENT_TEXT_PLAIN
代表：
public static final BasicProperties PERSISTENT_TEXT_PLAIN = new BasicProperties("text/plain",null,null,2,0, null, null, null,null, null, null, null,null, null);
//也可以通过这种方式设置；发送消息的参数设置 expiration过期时间 deliveryMode 消息持久化方式
AMQP.BasicProperties properties = new AMQP.BasicProperties()
.builder().expiration("10000").deliveryMode(2).build();
保证在服务器重启的时候可以保持不丢失相关信息，重点解决服务器的异常崩溃而导致的消息丢失问题。但是，将所有的消息都设置为持久化，会严重影响RabbitMQ的性能，写入硬盘的速度比写入内存的速度慢的不只一点点。对于可靠性不是那么高的消息可以不采用持久化处理以提高整体的吞吐率，在选择是否要将消息持久化时，需要在可靠性和吞吐量之间做一个权衡.
在上面的案例中，RabbitMQ 分发消息采用的轮训分发，但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中有个消费者A处理任务的速度非常快，而另外一个消费者B处理速度却很慢，这个时候我们还采用轮训分发的话就会发现消费者A早早的处理完后空闲在那，而消费者B还在处理，这时消费者A等待消费者B处理完任务后A消费者才会得到下一个任务消息；这就会浪费空闲消费者A发服务器资源；但RabbitMQ 并不知道这种情况它依然很公平的进行分发.
为了避免这种情况，我们可以设置参数 channel.basicQos(1),
意思就是说如果消费者对这个任务还没有处理完或者我还没有应答你，你先别分配给我，我目前只能处理一个任务，然后 rabbitmq 就会把该任务分配给没有那么忙的那个空闲消费者，当然如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的 worker（消费者服务）或者改变其它存储任务的策略.
说好听点就是不公平分发，其实它叫预取值，后面说明，预取值就是信道中可以允许未确认消息的最大值，如果是1，那处理快的就很快处理完可以处理下一条，慢的还得继续处理，不接受消息，实现不公平分发.
我们还需要设置手动应答 ，因为自动应答，会发现虽然实现不公平分发，但是还是一样的，每个消费者消费的数据量很大可能是一样的，因为自动应答是一旦发送到消费者代表完成，后续还会继续给这个消费者发送，但是手动应答则会发现，我消费的慢，会等消费者消费完才会被分配下一个消息处理；所以消费快的消费者会消费更多的消息.
消费者A消费者B代码改造:
public
class
ConsumerA {
//
工作队列名称
public
static
final
String QUEUE_NAME = "workQueue"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建队列 以防启动消费者发现队列不存在报错
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
//设置0代表轮询分发、1不公平分发、大于1代表预取值 channel.basicQos(1
);
System.out.println(
"消费者A(处理资源很快)开始监听队列消息...."
);
//
应答方式 true自动应答 false手动应答(若是手动应答必须设置false)
boolean
autoAck =
false
;
//
消费者消费消息requeue
channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) ->
{
try
{
//
这里我就一句打印语句，没有复杂逻辑，正常这里有复杂业务
System.out.println("A消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
Thread.sleep(
3000);
//
3秒才能处理完一个任务消息
//
手动确认应答 不批量应答
channel.basicAck(message.getEnvelope().getDeliveryTag(),
false
);
}
catch
(IOException |
InterruptedException e) {
e.printStackTrace();
//
出现异常手动进行不应答;并且放入队列中
channel.basicReject(message.getEnvelope().getDeliveryTag(),
true
);
}
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
//
消费者A(处理资源很快)开始监听队列消息....
//
A消费者获取队列信息并处理：这是一个编号为：0 的待处理的消息
//
A消费者获取队列信息并处理：这是一个编号为：2 的待处理的消息
//
A消费者获取队列信息并处理：这是一个编号为：3 的待处理的消息
//
A消费者获取队列信息并处理：这是一个编号为：4 的待处理的消息
//
A消费者获取队列信息并处理：这是一个编号为：6 的待处理的消息
//
A消费者获取队列信息并处理：这是一个编号为：7 的待处理的消息
//
A消费者获取队列信息并处理：这是一个编号为：8 的待处理的消息
public
class
ConsumerB {
//
工作队列名称
public
static
final
String QUEUE_NAME = "workQueue"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建队列 以防启动消费者发现队列不存在报错
channel.queueDeclare(QUEUE_NAME,
true
,
false
,
false
,
null
);
//设置0代表轮询分发、1不公平分发、大于1代表预取值 channel.basicQos(1
);
System.out.println(
"消费者B(处理资源很慢)开始监听队列消息...."
);
//
应答方式 true自动应答 false手动应答(若是手动应答必须设置false)
boolean
autoAck =
false
;
//
消费者消费消息requeue
channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, message) ->
{
try
{
//
这里我就一句打印语句，没有复杂逻辑，正常这里有复杂业务
System.out.println("B消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
Thread.sleep(
10000);
//
10秒才能处理完一个任务消息
//
手动确认应答 不批量应答
channel.basicAck(message.getEnvelope().getDeliveryTag(),
false
);
}
catch
(IOException |
InterruptedException e) {
e.printStackTrace();
//
出现异常手动进行不应答;并且放入队列中
channel.basicReject(message.getEnvelope().getDeliveryTag(),
true
);
}
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
//
消费者B(处理资源很慢)开始监听队列消息....
//
B消费者获取队列信息并处理：这是一个编号为：1 的待处理的消息
//
B消费者获取队列信息并处理：这是一个编号为：5 的待处理的消息
//
B消费者获取队列信息并处理：这是一个编号为：9 的待处理的消息
总结不公平分发就是，在消费者有收到确认机制后并设置不公平分发就代表哪个消费者先消费完后任务，RabbitMQ队列会先为它分配下一个任务消息，反之慢的消费者等消费完也可以拿到新消息处理 。
本身队列发送给消费者的消息是异步发送的，所以在任何时候，消费者连接队列时的channel上肯定不止一个消息,另外来自消费者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区，因此希望开发人员能 限制此缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用basicQos方法设置“预取计数”值来完成的。 该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，RabbitMQ 将停止再往通道上传递更多消息，除非至少有一个未处理的消息被确认后RabbitMQ才会再往信道上发送一条任务消息； 。
假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会再往该通道上再传递任何消息，除非至少有一个未应答的消息被ack。比方说tag=6这个消息刚刚被确认ACK，RabbitMQ将会感知这个tag=6被确认并再往信道发送一条消息.
消息应答和 QoS 预取值对用户吞吐量有重大影响。通常，增加预取将提高向消费者传递消息的速度。 虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的RAM消耗(随机存取存储器) ，应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同，100 到 300 范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的.
生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)， 一旦消息被投递到所匹配的队列之后，broker就会发送一个确认指令给生产者(包含消息的唯一ID) ，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列设置的是可持久化的，那么向生产者确认消息之前会先将消息写入磁盘之后再发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域（批量确认），表示当前序列号及这个序列号之前的所有消息都会一并确认.
confirm模式最大的好处在于它是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者程序同样可以在回调方法中处理该 nack 消息.
发布确认默认是没有开启的，如果要开启需要调用方法confirmSelect，每当你要想使用发布确认，都需要在channel上调用该方法 。
这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布者发布一个消息之后只有等RabbitMQ回调确认方法，发布者并且也接受到RabbitMQ的确认时，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。 这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了.
public
class
ProducerA {
//
单个发布确认
public
static
final
String SINGLE_RELEASE_CONFIRMATION = "singleReleaseConfirmation"
;
public
static
void
main(String[] args)
throws
IOException, InterruptedException {
long
begin = System.currentTimeMillis();
//
记录开始时间
//
获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建一个信道
channel.queueDeclare(SINGLE_RELEASE_CONFIRMATION,
true
,
false
,
false
,
null
);
//
开启发布确认功能
channel.confirmSelect();
//
循环发送消息
for
(
int
i = 0; i < 1000; i++
) {
String str
= "单个发布确认信息" +
i;
System.out.println(
"开始发送信息：" +
i);
//
发布信息
channel.basicPublish(""
, SINGLE_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
str.getBytes(StandardCharsets.UTF_8));
//
验证是否发送成功（等待确认）
//
channel.waitForConfirms(3000); 发送三秒后没得到回复将断定未发送过去
boolean b = channel.waitForConfirms();
if
(b) {
System.out.println(
"发送成功了：" +
i);
}
}
long
end = System.currentTimeMillis();
//
记录结尾时间
System.out.println("单个发布确认用时：" + (end - begin));
//单个发布确认用时：2278
}
}
与单个等待确认消息相比，先发布一批消息然后一起确认可以极大的提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道那一批确认的是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布.
public
class
ProducerA {
//
批量确认
public
static
final
String BATCH_CONFIRMATION = "batchConfirmation"
;
public
static
void
main(String[] args)
throws
IOException, InterruptedException {
long
begin = System.currentTimeMillis();
//
记录开始时间
//
获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建一个信道
channel.queueDeclare(BATCH_CONFIRMATION,
true
,
false
,
false
,
null
);
//
开启发布确认功能
channel.confirmSelect();
//
定义每次批量处理多少消息进行确认
int
batchNumber = 100
;
//
循环发送消息
for
(
int
i = 0; i < 1000; i++
) {
String str
= "批量发布确认信息" +
i;
System.out.println(
"开始发送信息：" +
i);
//
发布信息
channel.basicPublish(""
, BATCH_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
str.getBytes(StandardCharsets.UTF_8));
//
验证是否发送成功（等待确认） 用求余的方式来判断每轮100个
//
channel.waitForConfirms(3000); 发送三秒后没得到回复将断定未发送过去
if ((i + 1) % batchNumber == 0) {
if (
channel.waitForConfirms()
) {
System.out.println("批量发送成功了 范围为：" + (i - (batchNumber - 1)) + " ~ " +
i); } }
}
long
end = System.currentTimeMillis();
//
记录结尾时间
System.out.println("批量发布确认用时：" + (end - begin));
//批量发布确认用时：454
}
}
异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功，下面就让我们来详细讲解异步确认是怎么实现的.
public
class
ProducerA {
//
异步发布确认
public
static
final
String ASYNC_RELEASE_CONFIRMATION = "asyncReleaseConfirmation"
;
public
static
void
main(String[] args)
throws
IOException {
long
begin = System.currentTimeMillis();
//
记录开始时间
//
获取信道
Channel channel =
ChannelUtil.getChannel();
//
创建一个信道
channel.queueDeclare(ASYNC_RELEASE_CONFIRMATION,
true
,
false
,
false
,
null
);
//
开启发布确认功能
channel.confirmSelect();
//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Start
//
线程安全有序的一个哈希表Map，适用于高并发的情况
//
1.轻松的将序号与消息进行关联 2.轻松批量删除条目 只要给到序列号 3.支持并发访问
ConcurrentSkipListMap<Long, String> outstandingConfirms =
new
ConcurrentSkipListMap<>
();
ConfirmCallback ackCallback
= (deliveryTag, multiple) ->
{
//
这个是回调成功的，回调成功后把集合中的数据删除，最终就代表失败的多少
if
(multiple) {
ConcurrentNavigableMap
<Long, String> longStringConcurrentNavigableMap =
outstandingConfirms.headMap(deliveryTag,
true
);
longStringConcurrentNavigableMap.clear();
}
else
{
outstandingConfirms.remove(deliveryTag);
}
System.out.println(
"~~~~ 回调成功的数据：" + deliveryTag + " 是否批量确认：" +
multiple);
};
ConfirmCallback nackCallback
= (deliveryTag, multiple) ->
{
System.out.println(
"~~~~ 回调失败的数据：" +
deliveryTag);
};
//
添加监听器，监听返回(监听器一定要再发送消息之前就创建和监听) 参数1：回调成功 参数2：回调失败
channel.addConfirmListener(ackCallback, nackCallback);
//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ End
//
循环发送消息 （因为是异步 这里正常发送不用管它）
for
(
int
i = 0; i < 10000; i++
) {
String str
= "异步发布确认信息" +
i;
//
记录要发送的数据添加到集合中
outstandingConfirms.put(channel.getNextPublishSeqNo(),str);
//
发布信息
channel.basicPublish(""
, ASYNC_RELEASE_CONFIRMATION, MessageProperties.PERSISTENT_TEXT_PLAIN,
str.getBytes(StandardCharsets.UTF_8));
}
long
end = System.currentTimeMillis();
//
记录结尾时间
System.out.println("异步发布确认用时：" + (end - begin));
//
异步发布确认用时：337
}
}
交换机（Exchange）接收消息，并根据路由键(Routing Key)转发消息到绑定的队列 。
RabbitMQ消息传递模型的核心思想是: 生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列（需要有提前绑定路由键）还是说把它们群发到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定.
binding其实是exchange和queue之间的桥梁，它告诉我们exchange和那个队列进行了绑定关系.
直接(direct), 主题(topic) ,标题(headers) , 扇出(fanout) 。
每当我们消费者连接到RabbitMQ时，我们都需要一个全新的空队列（因为这个队列需要绑定到交换机上），为此我们可以创建一个具有随机名称的队列，或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接，队列将被自动删除.
我们没使用Exchange，但仍能够将消息发送到队列。之前能实现的原因是因为我们使用的是默认交换，我们通过空字符串("")进行标识
basicPublish( String exchange , String routingKey , BasicProperties props, byte[] body)
当初使用：channel.basicPublish( "" , "当初写队列名称" , null,"发送的消息");
第一个参数是交换机的名称： 空字符串表示默认或无名称交换机
消息能路由发送到队列中其实是由routingKey(binding key)绑定key指定的，那时key都填写队列名称，所有直接被绑定到对应队列，
可以说使用的是直接交换机（direct）
扇出交换机是最基本的交换机类型，它所能做的事情非常简单--- 广播消息 。扇出交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要“思考”，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的.
创建交换机方法：exchangeDeclare( String exchange , BuiltinExchangeType type , boolean durable , boolean autoDelete , boolean internal , Map<String, Object> arguments ) exchange:
交换机名称
type:
交换机类型，direct、topic、 fanout、 headers
durable:
是否需要持久化
autoDelete:
当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
internal:
当前Exchange是否用于RabbitMQ内部使用，默认为False
arguments:
扩展参数，用于扩展AMQP协议定制化使用
注：推荐在编写生产者时创建交换机，在编写消费者时应该创建队列，并且队列绑定交换机，启动时先启动交换机
public
class
FanoutConsumerA {
//
交换机名称
public
static
final
String FANOUT_DEMO_EXCHANGE = "fanoutDemoExchange"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置发布订阅模式（扇出模式）防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(FANOUT_DEMO_EXCHANGE, BuiltinExchangeType.FANOUT,
true
,
false
,
false
,
null
);
//
创建一个临时队列
String queueName =
channel.queueDeclare().getQueue();
//
把队列绑定到指定交换机上
channel.queueBind(queueName, FANOUT_DEMO_EXCHANGE, ""
);
//
接收队列消息
channel.basicConsume(queueName,
false
, (consumerTag, message) ->
{
System.out.println(
"A临时消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
FanoutConsumerB {
//
交换机名称
public
static
final
String FANOUT_DEMO_EXCHANGE = "fanoutDemoExchange"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置发布订阅模式（扇出模式）防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(FANOUT_DEMO_EXCHANGE, BuiltinExchangeType.FANOUT,
true
,
false
,
false
,
null
);
//
创建一个临时队列
String queueName =
channel.queueDeclare().getQueue();
//
把队列绑定到指定交换机上
channel.queueBind(queueName, FANOUT_DEMO_EXCHANGE, ""
);
//
接收队列消息
channel.basicConsume(queueName,
false
, (consumerTag, message) ->
{
System.out.println(
"B临时消费者获取队列信息并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
FanoutProducer {
//
交换机名称
public
static
final
String FANOUT_DEMO_EXCHANGE = "fanoutDemoExchange"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置发布订阅模式（扇出模式）
channel.exchangeDeclare(FANOUT_DEMO_EXCHANGE, BuiltinExchangeType.FANOUT,
true
,
false
,
false
,
null
);
//
循环发送消息
for
(
int
i = 0; i < 10; i++
) {
String str
= "异步发布确认信息" +
i;
//
发布信息
channel.basicPublish(FANOUT_DEMO_EXCHANGE, ""
, MessageProperties.PERSISTENT_TEXT_PLAIN,
str.getBytes(StandardCharsets.UTF_8));
}
System.out.println(
"消息发送完毕！"
);
}
}
直连交换机是一种带路由功能的交换机，一个队列会绑定到一个交换机上，一个交换机身上可以绑定多个队列；当生产者发送消息给交换机时，交换机会根据binding在交换机上的routing_key来查找路由，最终被送到指定的队列里；当一个交换机绑定多个队列，就会被送到对应的队列去处理.
下面我将以一个案例的方式来使用直接交换机，如下图：有一个 日志交换机（LogExchange） ，它负责的功能是将生产者发送的日志信息交到对应的队列中，队列分别为 基本日志队列（BasicLogQueue）、错误队列（ErrQueue）、通知队列（NotifyQueue） ； 其中基本日志队列记录日常运行日志错误队列记录重大问题信息，因为错误日志需要告知管理员，所有将错误日志又发送到通知队列来发送邮件告知 。
public
class
BasicLogConsumer {
//
交换机名称
public
static
final
String LOG_EXCHANGE = "LogExchange"
;
//
队列名称
public
static
final
String BASIC_LOG_QUEUE = "BasicLogQueue"
;
//
路由绑定关系 Routing Key
public
static
final
String BASIC_LOG_KEY = "BasicLogKey"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(LOG_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
创建一个基本日志队列
channel.queueDeclare(BASIC_LOG_QUEUE,
true
,
false
,
false
,
null
);
//
队列绑定到交换机上，并通过路由key来对应两者的连接
channel.queueBind(BASIC_LOG_QUEUE, LOG_EXCHANGE, BASIC_LOG_KEY);
//
接收队列消息
channel.basicConsume(BASIC_LOG_QUEUE,
true
, (consumerTag, message) ->
{
System.out.println(
"基本日志队列里获取的任务并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
ErrConsumer {
//
交换机名称
public
static
final
String LOG_EXCHANGE = "LogExchange"
;
//
队列名称
public
static
final
String ERR_QUEUE = "ErrQueue"
;
//
路由绑定关系 Routing Key
public
static
final
String ERR_KEY = "ErrKey"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(LOG_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
创建一个基本日志队列
channel.queueDeclare(ERR_QUEUE,
true
,
false
,
false
,
null
);
//
队列绑定到交换机上，并通过路由key来对应两者的连接
channel.queueBind(ERR_QUEUE, LOG_EXCHANGE, ERR_KEY);
//
接收队列消息
channel.basicConsume(ERR_QUEUE,
true
, (consumerTag, message) ->
{
System.out.println(
"错误日志队列里获取的任务并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
NotifyConsumer {
//
交换机名称
public
static
final
String LOG_EXCHANGE = "LogExchange"
;
//
队列名称
public
static
final
String NOTIFY_QUEUE = "NotifyQueue"
;
//
路由绑定关系 Routing Key
public
static
final
String ERR_KEY = "ErrKey"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(LOG_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
创建一个基本日志队列
channel.queueDeclare(NOTIFY_QUEUE,
true
,
false
,
false
,
null
);
//
队列绑定到交换机上，并通过路由key来对应两者的连接
channel.queueBind(NOTIFY_QUEUE, LOG_EXCHANGE, ERR_KEY);
//
接收队列消息
channel.basicConsume(NOTIFY_QUEUE,
true
, (consumerTag, message) ->
{
System.out.println(
"接收到错误日志并处理任务邮件发送，错误日志内容为：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
DirectProducer {
//
交换机名称
public
static
final
String LOG_EXCHANGE = "LogExchange"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(LOG_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
待发送的消息
HashMap<String, List<String>> sendMsg =
new
HashMap<>
();
List
<String> errMsg = Arrays.asList("[1001]系统存在重大问题，可能会发生宕机！！", "[1002]电脑受到蠕虫病毒攻击！！"
);
List
<String> basicMsg = Arrays.asList("[2001]尊敬的蚂蚁小哥欢迎登录系统", "[2002]蚂蚁小哥已退出账号"
);
sendMsg.put(
"ErrKey"
, errMsg);
sendMsg.put(
"BasicLogKey"
, basicMsg);
//
循环发送消息任务
for
(Map.Entry<String, List<String>>
msg : sendMsg.entrySet()) {
String key
= msg.getKey();
//
路由key
List<String> messages = msg.getValue();
//
待发送消息
for
(String message : messages) {
channel.basicPublish(LOG_EXCHANGE, key, MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes(StandardCharsets.UTF_8));
}
}
System.out.println(
"消息发送完成！！！"
);
}
}
从上面可以看出若exchange的绑定类型是direct，但是它 绑定的多个队列的key如果都相同，在这种情况下虽然绑定类型是direct但是它表现的就和fanout有点类似了 ，就跟广播差不多.
适用场景：有优先级的任务，根据任务的优先级把消息发送到对应的队列，这样可以指派更多的资源去处理高优先级的队列.
之前我们使用只能进行随意广播的fanout扇出交换机，但只能群发给每个队列，不能发送到指定某个队列，但是使用了direct交换机，就可以实现有选择性地发送到指定队列了。尽管使用direct交换机，但是它仍然存在局限性，如果我们希望一条消息发送给多个队列，那么这个交换机需要绑定上非常多的routing_key，假设每个交换机上都绑定一堆的routing_key连接到各个队列上。那么消息的管理就会异常地困难。所以RabbitMQ提供了一种主题交换机，发送到主题交换机上的消息需要携带指定规则的routing_key，主题交换机会根据这个规则将数据发送到对应的(多个)队列上；可以理解为模糊匹配.
主题交换机的routing_key需要有一定的规则，交换机和队列的binding_key需要采用*.#.*.....的格式，每个部分用 . 分开。 其中： * (星号)： 可以代替一个单词 #(井号)： 可以替代零个或多个单词 。
上图是一个队列绑定关系图，我们来看看他们之间数据接收情况是怎么样的：
队列绑定交换机的Key 匹配规则 quick . orange . rabbit
被队列 Q1 Q2 接收到
lazy . orange . elephant
被队列 Q1 Q2 接收到
quick . orange . fox
被队列 Q1 接收到
lazy . brown . fox
被队列 Q2 接收到
lazy . pink . rabbit
虽然满足两个绑定规则但两个规则都是在Q2队列，所有只有Q2接收一次
quick . brown . fox
不匹配任何绑定不会被任何队列接收到会被丢弃
quick . orange . male . rabbit
是四个单词不匹配任何绑定会被丢弃
lazy . orange . male . rabbit
是四个单词但匹配 Q2
注：当一个队列绑定键是#,那么这个队列将接收所有数据，就有点像fanout扇出交换机了；如果队列绑定键当中没有#和*出现，那么该队列绑定类型就是direct直接交换机了； 下面我将以上图的案例来代码实现:
public
class
CAConsumer {
//
交换机名称
public
static
final
String TOPIC_EXCHANGE = "TopicExchange"
;
//
队列Q1名称
public
static
final
String Q1 = "Q1Queue"
;
//
路由绑定关系 Routing Key
public
static
final
String Q1_KEY = "*.orange.*"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为主题交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC,
true
,
false
,
false
,
null
);
//
创建一个 Q1 队列
channel.queueDeclare(Q1,
true
,
false
,
false
,
null
);
//
队列绑定到交换机上，并通过主题路由key来对应两者的连接
channel.queueBind(Q1, TOPIC_EXCHANGE, Q1_KEY);
//
接收队列消息
channel.basicConsume(Q1,
true
, (consumerTag, message) ->
{
System.out.println(
"Q1获取的任务并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
CBConsumer {
//
交换机名称
public
static
final
String TOPIC_EXCHANGE = "TopicExchange"
;
//
队列Q2名称
public
static
final
String Q2 = "Q2Queue"
;
//
路由绑定关系 Routing Key 1
public
static
final
String Q2_KEY_A = "*.*.rabbit"
;
//
路由绑定关系 Routing Key 2
public
static
final
String Q2_KEY_B = "lazy.#"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为主题交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC,
true
,
false
,
false
,
null
);
//
创建一个 Q2 队列
channel.queueDeclare(Q2,
true
,
false
,
false
,
null
);
//
队列绑定到交换机上，并通过主题路由key来对应两者的连接(这里设置了2个连接)
channel.queueBind(Q2, TOPIC_EXCHANGE, Q2_KEY_A);
channel.queueBind(Q2, TOPIC_EXCHANGE, Q2_KEY_B);
//
接收队列消息
channel.basicConsume(Q2,
true
, (consumerTag, message) ->
{
System.out.println(
"Q2获取的任务并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
public
class
TopicProducer {
//
交换机名称
public
static
final
String TOPIC_EXCHANGE = "TopicExchange"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明exchange交换机 并设置为主题交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC,
true
,
false
,
false
,
null
);
//
消息任务准备
HashMap<String, String> sendMsg =
new
HashMap<>
();
sendMsg.put(
"quick.orange.rabbit", "被队列 Q1 Q2 接收到"
);
sendMsg.put(
"lazy.orange.elephant", "被队列 Q1 Q2 接收到"
);
sendMsg.put(
"quick.orange.fox", "被队列 Q1 接收到"
);
sendMsg.put(
"lazy.brown.fox", "被队列 Q2 接收到"
);
sendMsg.put(
"lazy.pink.rabbit", "虽然满足两个绑定规则但两个规则都是在Q2队列，所有只要Q2接收一次"
);
sendMsg.put(
"quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃"
);
sendMsg.put(
"quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃"
);
sendMsg.put(
"lazy.orange.male.rabbit", "是四个单词但匹配 Q2"
);
//
循环发送消息任务
for
(Map.Entry<String, String>
msg : sendMsg.entrySet()) {
String routKey
= msg.getKey();
//
主题路由key
String message = msg.getValue();
//
消息任务
channel.basicPublish(TOPIC_EXCHANGE, routKey, MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes(StandardCharsets.UTF_8));
}
}
}
死信队列 (DLX,Dead-Letter-Exchange) 就是无法被消费的消息，一般来说producer将消息投递到broker或者直接到queue里了，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。当消息在一个队列中变成死信后，它能被重新发布到另一个Exchange中，通过Exchange分发到另外的队列；本质就是该消息不会再被任何消费端消费（但你可以自定义某消费者单独处理这些死信）。 应用场景:为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ的死信队列机制，当消息消费发生异常时，将消息投入死信队列中；比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效 。
“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃.
①：消息被拒绝（basic.reject/basic.nack)，且 requeue = false(代表不重新回到队列) ②：消息因TTL过期（就是任务消息上携带过期时间） ③：消息队列的消息数量已经超过最大队列长度， 先入队的消息会被丢弃变为死信
普通消费者代码:
public
class
TTLConsumer {
//
声明普通的交换机名称
public
static
final
String NORMAL_EXCHANGE = "NormalExchange"
;
//
声明死信交换机名称
public
static
final
String DLX_EXCHANGE = "DLXExchange"
;
//
声明普通队列名称
public
static
final
String Normal_Queue = "NormalQueue"
;
//
声明死信队列名称
public
static
final
String DLX_QUEUE = "DLXQueue"
;
//
声明路由绑定关系 Routing Key 普通交换机到普通队列
public
static
final
String NORMAL_KEY = "NormalKey"
;
//
声明路由绑定关系 Routing Key 死信交换机到死信队列
public
static
final
String DLX_KEY = "DLXKey"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明普通exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
声明死信exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
声明死信队列
channel.queueDeclare(DLX_QUEUE,
true
,
false
,
false
,
null
);
//
死信队列绑定死信交换机routingKey
channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_KEY);
//
参数设置
Map<String, Object> arguments =
new
HashMap<>
();
//
正常队列设置死信交换机 参数key是固定值
arguments.put(" x-dead-letter-exchange ", DLX_EXCHANGE);
//
正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值
arguments.put(" x-dead-letter-routing-key ", DLX_KEY);
//
声明普通队列
channel.queueDeclare(Normal_Queue,
true
,
false
,
false
, arguments);
//
普通队列绑定普通交换机routingKey
channel.queueBind(Normal_Queue, NORMAL_EXCHANGE, NORMAL_KEY);
System.out.println(
"初始化完成，等待接收消息"
);
//
接收队列消息
channel.basicConsume(Normal_Queue,
true
, (consumerTag, message) ->
{
System.out.println(
"如同队列获取的任务并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
死信消费者代码:
public
class
DLXConsumer {
//
声明死信交换机名称
public
static
final
String DLX_EXCHANGE = "DLXExchange"
;
//
声明死信队列名称
public
static
final
String DLX_QUEUE = "DLXQueue"
;
//
声明路由绑定关系 Routing Key 死信交换机到死信队列
public
static
final
String DLX_KEY = "DLXKey"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明死信exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
声明死信队列
channel.queueDeclare(DLX_QUEUE,
true
,
false
,
false
,
null
);
//
死信队列绑定死信交换机routingKey
channel.queueBind(DLX_QUEUE, DLX_EXCHANGE, DLX_KEY);
System.out.println(
"初始化完成，等待接收消息"
);
//
接收队列消息
channel.basicConsume(DLX_QUEUE,
true
, (consumerTag, message) ->
{
System.out.println(
"死信队列里获取的任务并处理：" +
new
String(message.getBody(), StandardCharsets.UTF_8));
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
}
}
生产者代码编写:
public
class
DLXProducer {
//
声明普通的交换机名称
public
static
final
String NORMAL_EXCHANGE = "NormalExchange"
;
//
声明路由绑定关系 Routing Key 普通交换机到普通队列
public
static
final
String NORMAL_KEY = "NormalKey"
;
public
static
void
main(String[] args)
throws
IOException {
//
调用自己的工具类获取信道
Channel channel =
ChannelUtil.getChannel();
//
声明普通exchange交换机 并设置为直接交换机；防止消费者先启动报错，找不到交换机
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
true
,
false
,
false
,
null
);
//
发送消息的参数设置 expiration过期时间10秒 deliveryMode 消息持久化方式
AMQP.BasicProperties properties =
new
AMQP.BasicProperties()
.builder()
.expiration(
"10000")
.deliveryMode(2
).build();
//
循环发送消息
for
(
int
i = 0; i < 5; i++
) {
String str
= "测试队列任务 " +
i;
//
发布信息
channel.basicPublish(NORMAL_EXCHANGE, NORMAL_KEY, properties,
str.getBytes(StandardCharsets.UTF_8));
}
System.out.println(
"消息发送完毕！"
);
}
}
编写好代码以后执行普通消费者和死信消费者代码查看创建的队列及交换机的状况 。
测试（为了可以更好的演示效果，关闭 普通消费者 和 死信消费者 ）:
生产者发送5条消息到普通 队列中，此时普通队列里面存在10条未消费信息:
消息达到过期时间后会从普通队列推送到死信队列里（因为提前设置了消息变死信后发送到死信交换机） 。
接下来我们就可以启动死信消费者来消费这一批死信队列里的任务消息 。
代码优化：剔除生产者代码中的消息过期时间，并在普通消费者里面设置队列最大长度 。
//
参数设置
Map<String, Object> arguments =
new
HashMap<>
();
//
正常队列设置死信交换机 参数key是固定值
arguments.put("x-dead-letter-exchange"
, DLX_EXCHANGE);
//
正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值
arguments.put("x-dead-letter-routing-key"
, DLX_KEY);
//设置正常队列的长度限制 为3 arguments.put("x-max-length",3);
注：因为队列参数改变，需要先删除原队列，并启动消费者，创建出带队列长度的队列 。
代码优化：剔除普通消费者里面设置队列最大长度，并优化普通消费者消息接收代码 。
//
接收队列消息
channel.basicConsume(Normal_Queue,
false
, (consumerTag, message) ->
{
//
获取的任务消息
String msg =
new
String(message.getBody(), StandardCharsets.UTF_8);
//
手动不确认，拒收，并丢去队列
if
("测试队列任务 3"
.equals(msg)) {
//出现异常手动进行不应答;并且不放入队列中 channel.basicReject(message.getEnvelope().getDeliveryTag(), false
);
}
else
{
System.out.println(
"如同队列获取的任务并处理：" +
msg);
}
}, consumerTag
->
{
System.out.println(
"监听的队列出现异常；可能队列被删除！"
);
});
普通队列： 它是一种队列，队列意味着内部的元素是有序的，元素出队和入队是有方向性的，元素从一端进入，从另一端取出.
延时队列： 最重要的特性就体现在它的延时属性上，跟普通的队列不一样的是，普通队列中的元素总是等着希望被早点取出处理，而 延时队列中的元素则是希望被在指定时间得到取出和处理 ，所以延时队列中的元素是都是带时间属性的，通常来说是需要被处理的消息或者任务。简单来说，延时队列就是用来存放需要在指定时间以后被处理的元素的队列（到达设置的延迟时间后再推给消费者进行任务处理）.
那么什么时候需要用延时队列呢？考虑一下以下场景：
①：订单在十分钟之内未支付则自动取消。
②：新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
③：账单在一周内未支付，则自动结算。
④：用户注册成功后，如果三天内没有登陆则进行短信提醒。
⑤：用户发起退款，如果三天内没有得到处理则通知相关运营人员。
⑥：预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。
TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了TTL属性或者进入了设置TTL属性的队列，那么这条消息如果在TTL设置的时间内没有被消费，则会成为“死信”。如果同时配置了队列的TTL和消息的TTL，那么较小的那个值将会被使用.
设置这个TTL值有两种方式（队列设置、消息设置）：
第一种是在创建队列的时候设置队列的 "x-message-ttl" 属性，如下：
Map<String, Object> arguments = new HashMap<>(); //设置消息延迟10秒；投递到该队列的消息超过10秒直接丢弃 arguments.put("x-message-ttl",10000); //创建队列，并指定参数 channel.queueDeclare(Normal_Queue, true, false, false, arguments); 第二种方式针对每条消息设置TTL：
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); //这条消息的过期时间也被设置成了10s , 超过10秒未处理则执行到此消息后被丢弃 builder.expiration("10000"); AMQP.BasicProperties properties = builder.build(); channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes()); 区别的： 第一种在队列上设置TTL属性，那么 一旦消息过期，就会被队列丢弃 ；而第二种方式，消息即使过期，也不一定会被马上丢弃， 因为 消息是否过期是在即将投递到消费者之前判定的 ，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间。 另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，
否则该消息将会被丢弃。
看到这里也代表基本的RabbitMQ已经知道了，下面可以看一看下篇的 SpringBoot整合RabbitMQ ，下篇有延迟队列的详细说明.
最后此篇关于RabbitMQ详解（上）的文章就讲到这里了,如果你想了解更多关于RabbitMQ详解（上）的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
30
4
0
我想读取 RabbitMQ 队列中未确认消息的负载或 messageId。这可能吗？ 我想这样做的原因是我尝试使用 RabbitMQ 死信功能来构建一个循环以定期自动生成消息。简而言之，创建两个队列
除了 vFabric 由 VMWare 提供商业支持之外，vFabric RabbitMQ 和 RabbitMQ 之间的主要区别是什么？ 最佳答案 来自 source : We also produc
RabbitMQ 集群中有如下三个节点。 在 RabbitMQ 中，有两个队列，q1 和 q2。 q1 和q2 的主副本分布在不同的节点上。两个队列都被其他节点镜像。 三个节点前面有一个负载均衡器。
我希望在谷歌计算引擎上实现 rabbitmq 来处理我的 android 和 ios 消息传递应用程序上的消息。我听说 rabbitmq 可能非常耗电，所以我想知道解决这个问题的最佳解决方案是什么？我
是否可以在 RabbitMQ 服务器(管理插件 View )中查看连接单元的主机名，而不仅仅是 IP/端口？我们使用动态连接位置，这样更容易识别客户...... 最佳答案 不，没有这样的内置功能。 作
我正在阅读 RabbitMQ in Action 书，仍在第 2 章中，但作者说的一件事让我感到困惑。您设置了一个交换并发送了一条消息，两个订阅者正在监听队列。当第一条消息进来时，第一个订阅者得到它，
我正在使用 RabbitMQ 将所有消息排队，并将消息作为 SMS 发送给各个消费者。我正在使用直接交换，并且我已经正确地创建了一个到带有路由键的队列的绑定(bind)。问题是，当我尝试发布消息时，我
我们正在使用微服务架构在 nodejs 中实现 Web-API。每个服务都会公开 HTTP 端点，以便应用程序/网站可以与其交互。为了同步不同的数据库，我们目前使用 RabbitMQ。微服务可以在扇出
我计划在 RabbitMQ 消息头中存储堆栈跟踪。消息 header 是否有大小限制？ 最佳答案 RabbitMQ 默认使用 AMQP 版本 0.9.1。根据AMQP protocol specifi
无法理解 exclusive queue 和 exclusive consumer 之间的区别，想知道我是否理解正确。 假设我有一个 queue、consumer1 和 consumer2。 我的理解
发布到 RabbitMQ 队列(发布/订阅模型)时消息的最大大小是多少？ 我在文档中看不到任何明确的限制，但我认为有一些指导方针。 提前致谢。 最佳答案 我在做比较亚马逊队列服务和 RabbitMQ
我可以使用 Publish/Subscribe 创建扇出交换RabbitMQ Java 教程，任何连接的消费者都会收到一条消息的副本。我不想以动态/编程方式声明交换和绑定(bind)，而是想在连接任何
java的 native rabbitmq客户端允许在连接设置上设置心跳，例如: import com.rabbitmq.client.ConnectionFactory; ... Connectio
我开始着手一个新项目，我们被要求将系统构建为一系列微服务，使用 RabbitMQ 作为它们之间的通信层。 在开发 REST API 时，我倾向于使用接受 HTTP header 来控制版本控制，我看到
在 Rabbit MQ 中使用集群时，我计划使用竞争订阅者模式。 Producer : 1 Exchange : 1 direct Queue : 1 Consumers : n (multiple)
是否可以实现 aggregator pattern在 RabbitMQ 中？ 我有 A … N在发送到另一个队列之前我需要等待/聚合的消息 X . 所以我想我会有一些 唯一 ID 确保消息被路由 独家
我正在使用RabbitMQ向用户发送通知。用户可以随时读取其队列。 我面临的问题是，队列在夜间充满了很多通知，而当用户在早上返回时，他必须顺序处理这些消息。这些通知中有很多甚至是重复的。 我认为在发布
是否可以延迟通过 RabbitMQ 发送消息？ 例如，我想在 30 分钟后使客户端 session 过期，并且我发送了一条将在 30 分钟后处理的消息。 最佳答案 您可以尝试两种方法: 旧方法:在每个
RabbitMQ 在单个服务器上可以处理的最大队列数是多少？ 这取决于内存吗？它取决于 erlang 进程吗？ 最佳答案 RabbitMQ 代理内部没有任何硬编码限制。代理将利用所有可用资源(除非您对
RabbitMQ Server 使用或需要在防火墙上为节点集群打开哪些端口？ 我的 /usr/lib/rabbitmq/bin/rabbitmq-env 设置如下，我假设需要 (35197)。 SER
我是一名优秀的程序员,十分优秀！