逆水行舟用力撑，一篙松劲退千寻。这篇文章主要讲述RabbitMQ-进阶相关的知识，希望能为你提供帮助。
过期时间TTL过期时间TTL表示可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；
过了之后消息将自动被删除。
RabbitMQ可以对 消息和队列 设置TTL。目前有两种方法可以设置。
- 第一种方法是通过队列属性设置，队列中所有消息都有相同的过期时间。
- 第二种方法是对消息进行单独设置，每条消息TTL可以不同。
消息在队列的生存时间一旦超过设置的TTL值，就称为dead message被投递到死信队列， 消费者将无法再收到该消息。
设置队列TTL
//设置相关的配置，也可以在Web界面中设置
package com.zwt.springbootfanout.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
@Configuration
public class TTLRabbitMqConfiguration {
@Bean
public DirectExchange ttlDirectExchange() {return new DirectExchange("ttl_direct_exchange", true, true);
}@Bean
public Queue tttDireclQueue() {HashMap<
String, Integer>
map = new HashMap<
>
();
map.put("x-message-ttl",2000);
return new Queue("ttl.direct.queue",true);
}@Bean
public Binding ttlDirectBinding() {
return BindingBuilder.bind(tttDireclQueue()).to(ttlDirectExchange()).with("ttl");
}
}
expiration 字段以微秒为单位表示 TTL 值。
且与 x-message-ttl 具有相同的约束条件。
因为 expiration 字段必须为字符串类型，broker 将只会接受以字符串形式表达的数字。
当同时指定了 queue 和 message 的 TTL 值，则两者中较小的那个才会起作用。
消息确认机制的配置NONE值是禁用发布确认模式，是默认值
CORRELATED值是发布消息成功到交换器后会触发回调方法
SIMPLE值经测试有两种效果
其一效果和CORRELATED值一样会触发回调方法，
其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，
根据返回结果来判定下一步的逻辑，
要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker; < /pre>
死信队列DLX，全称为 Dead-Letter-Exchange , 可以称之为死信交换机，也有人称之为死信邮箱。
当消息在一个队列中变成死信(dead message)之后，它能被重新发送到另一个交换机中，这个交换机就是DLX ，绑定DLX的队列就称之为死信队列。
消息变成死信，可能是由于以下的原因：
- 消息被拒绝
- 消息过期
- 队列达到最大长度
当这个队列中存在死信时，Rabbitmq就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。
要想使用死信队列，只需要在定义队列的时候设置队列参数 x-dead-letter-exchange 指定交换机即可。
文章图片
内存磁盘的监控当内存使用超过配置的阈值或者磁盘空间剩余空间对于配置的阈值时，
RabbitMQ会暂时阻塞客户端的连接，并且停止接收从客户端发来的消息，以此避免服务器的崩溃，
客户端与服务端的心态检测机制也会失效。
当出现blocking或blocked话说明到达了阈值和以及高负荷运行了。
RabbitMQ的内存控制当出现警告的时候，可以通过配置去修改和调整。
命令的方式
rabbitmqctl set_vm_memory_high_watermark <
fraction>
rabbitmqctl set_vm_memory_high_watermark absolute 50MB
fraction/value 为内存阈值。
默认情况是：0.4/2GB，代表的含义是：当RabbitMQ的内存超过40%时，就会产生警告并且阻塞所有生产者的连接。
通过此命令修改阈值在Broker重启以后将会失效，通过修改配置文件方式设置的阈值则不会随着重启而消失，但修改了配置文件一样要重启broker才会生效。
rabbitmqctl set_vm_memory_high_watermark absolute 50MB
配置文件方式 rabbitmq.conf当前配置文件：/etc/rabbitmq/rabbitmq.conf
#默认
#vm_memory_high_watermark.relative = 0.4
# 使用relative相对值进行设置fraction,建议取值在04~0.7之间，不建议超过0.7.
vm_memory_high_watermark.relative = 0.6
# 使用absolute的绝对值的方式，但是是KB,MB,GB对应的命令如下
vm_memory_high_watermark.absolute = 2GB<
RabbitMQ的内存换页在某个Broker节点及内存阻塞生产者之前，它会尝试将队列中的消息换页到磁盘以释放内存空间，
持久化和非持久化的消息都会写入磁盘中，
其中持久化的消息本身就在磁盘中有一个副本，所以在转移的过程中持久化的消息会先从内存中清除掉。
默认情况下，内存到达的阈值是50%时就会换页处理。
也就是说，在默认情况下该内存的阈值是0.4的情况下，当内存超过0.4*0.5=0.2时，会进行换页动作。
比如有1000MB内存，当内存的使用率达到了400MB,
已经达到了极限，但是因为配置的换页内存0.5，这个时候会在达到极限400mb之前，会把内存中的200MB进行转移到磁盘中。从而达到稳健的运行。
可以通过设置 vm_memory_high_watermark_paging_ratio 来进行调整。
RabbitMQ的磁盘预警当磁盘的剩余空间低于确定的阈值时，RabbitMQ同样会阻塞生产者，
这样可以避免因非持久化的消息持续换页而耗尽磁盘空间导致服务器崩溃。
默认情况下：磁盘预警为50MB的时候会进行预警。
表示当前磁盘空间第50MB的时候会阻塞生产者并且停止内存消息换页到磁盘的过程。
通过命令方式修改如下：
rabbitmqctl set_disk_free_limit<
disk_limit>
rabbitmqctl set_disk_free_limit memory_limit<
fraction>
disk_limit：固定单位 KB MB GB
fraction ：是相对阈值，建议范围在:1.0~2.0之间。（相对于内存）
通过配置文件配置如下：
disk_free_limit.relative = 3.0
disk_free_limit.absolute = 50mb
集群 RabbitMQ 集群RabbitMQ这款消息队列中间件产品本身是基于Erlang编写，Erlang语言天生具备分布式特性（通过同步Erlang集群各节点的magic cookie来实现）。
因此，RabbitMQ天然支持Clustering。
这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过ZooKeeper分别来实现HA方案和保存集群的元数据。
集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。
主要参考官方文档： https://www.rabbitmq.com/clustering.html
集群搭建
ps aux|grep rabbitmqsystemctl status rabbitmq-server场景：假设有两个rabbitmq节点，分别为rabbit-1, rabbit-2，rabbit-1作为主节点，rabbit-2作为从节点。启动命令：RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server -detached结束命令：rabbitmqctl -n rabbit-1 stop第一步：启动第一个节点rabbit-1>
sudo RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server start &
...............省略...................
##########Logs: /var/log/rabbitmq/rabbit-1.log
########/var/log/rabbitmq/rabbit-1-sasl.log
##########
Starting broker...
completed with 7 plugins.启动第二个节点rabbit-2 注意：web管理插件端口占用,所以还要指定其web插件占用的端口号RABBITMQ_SERVER_START_ARGS=”-rabbitmq_management listener [{port,15673}]”sudo RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server start &
..............省略..................
##########Logs: /var/log/rabbitmq/rabbit-2.log
########/var/log/rabbitmq/rabbit-2-sasl.log
##########
Starting broker...
completed with 7 plugins. 验证启动 “ps aux|grep rabbitmq” rabbit-1操作作为主节点 #停止应用
>
sudo rabbitmqctl -n rabbit-1 stop_app
#目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
>
sudo rabbitmqctl -n rabbit-1 reset
#启动应用
>
sudo rabbitmqctl -n rabbit-1 start_apprabbit2操作为从节点# 停止应用
>
sudo rabbitmqctl -n rabbit-2 stop_app
# 目的是清除节点上的历史数据（如果不清除，无法将节点加入到集群）
>
sudo rabbitmqctl -n rabbit-2 reset
# 将rabbit2节点加入到rabbit1（主节点）集群当中【Server-node服务器的主机名】
>
sudo rabbitmqctl -n rabbit-2 join_cluster rabbit-1@\'Server-node\'
# 启动应用
>
sudo rabbitmqctl -n rabbit-2 start_app验证集群状态
>
sudo rabbitmqctl cluster_status -n rabbit-1
//集群有两个节点：rabbit-1@Server-node、rabbit-2@Server-node
[{nodes,[{disc,[\'rabbit-1@Server-node\',\'rabbit-2@Server-node\']}]},
{running_nodes,[\'rabbit-2@Server-node\',\'rabbit-1@Server-node\']},
{cluster_name,<
<
"rabbit-1@Server-node.localdomain">
>
},
{partitions,[]},
{alarms,[{\'rabbit-2@Server-node\',[]},{\'rabbit-1@Server-node\',[]}]}] 注意在访问的时候：web结面的管理需要给15672 node-1 和15673的node-2 设置用户名和密码。rabbitmqctl -n rabbit-1 add_user admin admin
rabbitmqctl -n rabbit-1 set_user_tags admin administrator
rabbitmqctl -n rabbit-1 set_permissions -p / admin ".*" ".*" ".*"
rabbitmqctl -n rabbit-2 add_user admin admin
rabbitmqctl -n rabbit-2 set_user_tags admin administrator
rabbitmqctl -n rabbit-2 set_permissions -p / admin ".*" ".*" ".*"
Tips：
如果采用多机部署方式，需读取其中一个节点的cookie, 并复制到其他节点（节点之间通过cookie确定相互是否可通信）。
cookie存放在/var/lib/rabbitmq/.erlang.cookie。
例如：主机名分别为rabbit-1、rabbit-2
1、逐个启动各节点
2、配置各节点的hosts文件( vim /etc/hosts)
ip1：rabbit-1
ip2：rabbit-2
其它步骤雷同单机部署方式
分布式事务分布式事务指事务的操作位于不同的节点上，需要保证事务的 AICD 特性。
例如：在下单场景下，库存和订单如果不在同一个节点上，就涉及分布式事务。
分布式事务的方式在分布式系统中，要实现分布式事务，无外乎那几种解决方案。
两阶段提交（2PC）需要数据库产商的支持，java组件有atomikos等两阶段提交（Two-phase Commit，2PC），通过引入协调者（Coordinator）来协调参与者的行为，并最终决定这些参与者是否要真正执行事务。
准备阶段协调者询问参与者事务是否执行成功，参与者发回事务执行结果。
文章图片
提交阶段如果事务在每个参与者上都执行成功，事务协调者发送通知让参与者提交事务；
否则，协调者发送通知让参与者回滚事务。
需要注意的是，在准备阶段，参与者执行了事务，但是还未提交。只有在提交阶段接收到协调者发来的通知后，才进行提交或者回滚。
存在的问题
- 同步阻塞 所有事务参与者在等待其它参与者响应的时候都处于同步阻塞状态，无法进行其它操作。
- 单点问题 协调者在 2PC 中起到非常大的作用，发生故障将会造成很大影响。特别是在阶段二发生故障，所有参与者会一直等待状态，无法完成其它操作。
- 数据不一致 在阶段二，如果协调者只发送了部分 Commit 消息，此时网络发生异常，那么只有部分参与者接收到 Commit 消息，也就是说只有部分参与者提交了事务，使得系统数据不一致。
- 太过保守 任意一个节点失败就会导致整个事务失败，没有完善的容错机制。
它分为三个阶段：
- Try 阶段主要是对业务系统做检测及资源预留。
- Confirm 阶段主要是对业务系统做确认提交，Try阶段执行成功并开始执行 Confirm阶段时，默认 - - - Confirm阶段是不会出错的。即：只要Try成功，Confirm一定成功。
- Cancel 阶段主要是在业务执行错误，需要回滚的状态下执行的业务取消，预留资源释放。
思路大概是： 我们有一个本地方法，里面依次调用
1：首先在 Try 阶段，要先调用远程接口把 Smith 和 Bob 的钱给冻结起来。
2：在 Confirm 阶段，执行远程调用的转账的操作，转账成功进行解冻。
3：如果第2步执行成功，那么转账成功，如果第二步执行失败，则调用远程冻结接口对应的解冻方法 (Cancel)。
优点： 跟2PC比起来，实现以及流程相对简单了一些，但数据的一致性比2PC也要差一些
缺点： 缺点还是比较明显的，在2,3步中都有可能失败。
TCC属于应用层的一种补偿方式，所以需要程序员在实现的时候多写很多补偿的代码，在一些场景中，一些业务流程可能用TCC不太好定义及处理。
本地消息表（异步确保）比如：支付宝、微信支付主动查询支付状态，对账单的形式本地消息表与业务数据表处于同一个数据库中，这样就能利用本地事务来保证在对这两个表的操作满足事务特性，并且使用了消息队列来保证最终一致性。
- 在分布式事务操作的一方完成写业务数据的操作之后向本地消息表发送一个消息，本地事务能保证这个消息一定会被写入本地消息表中。
- 之后将本地消息表中的消息转发到 Kafka 等消息队列中，如果转发成功则将消息从本地消息表中删除，否则继续重新转发。
- 在分布式事务操作的另一方从消息队列中读取一个消息，并执行消息中的操作。
缺点： 消息表会耦合到业务系统中，如果没有封装好的解决方案，会有很多杂活需要处理。
MQ 事务消息 异步场景，通用性较强，拓展性较高有一些第三方的MQ是支持事务消息的，比如RocketMQ，
他们支持事务消息的方式也是类似于采用的二阶段提交，
但是市面上一些主流的MQ都是不支持事务消息的，比如 Kafka 不支持。
以阿里的 RabbitMQ 中间件为例，其思路大致为：
- 第一阶段Prepared消息，会拿到消息的地址。
- 第二阶段执行本地事务
- 第三阶段通过第一阶段拿到的地址去访问消息，并修改状态。
缺点： 实现难度大，主流MQ不支持，RocketMQ事务消息部分代码也未开源。
通过本文我们总结并对比了几种分布式分解方案的优缺点，分布式事务本身是一个技术难题，
是没有一种完美的方案应对所有场景的，具体还是要根据业务场景去抉择吧。
阿里RocketMQ去实现的分布式事务，现在也有除了很多分布式事务的协调器，比如LCN等，大家可以多去尝试。
具体实现 分布式事务的完整架构图
文章图片
美团外卖的架构
文章图片
系统与系统之间的分布式事务问题
文章图片
系统间调用过程中事务回滚问题
import com.zwt.rabbitmq.dao.OrderDataBaseService;
import com.zwt.rabbitmq.pojo.Order;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
@Service
public class OrderService {
@Autowired
private OrderDataBaseService orderDataBaseService;
// 创建订单
@Transactional(rollbackFor = Exception.class) // 订单创建整个方法添加事务
public void createOrder(Order orderInfo) throws Exception {
// 1: 订单信息--插入丁订单系统，订单数据库事务
orderDataBaseService.saveOrder(orderInfo);
// 2：通过Http接口发送订单信息到运单系统
String result = dispatchHttpApi(orderInfo.getOrderId());
if(!"success".equals(result)) {
throw new Exception("订单创建失败,原因是运单接口调用失败!");
}
}
/**
*模拟http请求接口发送，运单系统，将订单号传过去 springcloud
* @return
*/
private String dispatchHttpApi(String orderId) {
SimpleClientHttpRequestFactory factory= new SimpleClientHttpRequestFactory();
// 链接超时 >
3秒
factory.setConnectTimeout(3000);
// 处理超时 >
2秒
factory.setReadTimeout(2000);
// 发送http请求
String url = "http://localhost:9000/dispatch/order?orderId="+orderId;
RestTemplate restTemplate = new RestTemplate(factory);
//异常
String result = restTemplate.getForObject(url, String.class);
return result;
}
}
基于MQ的分布式事务消息的可靠生产问题
文章图片
如果这个时候MQ服务器出现了异常和故障，那么消息是无法获取到回执信息。
如何是好？
基于MQ的分布式事务消息的可靠生产问题-定时重发如果出现异常，咱们就重发消息。
基于MQ的分布式事务消息的可靠消费
文章图片
基于MQ的分布式事务消息的消息重发设置重试次数二定要进行控制或者 try/catch
基于MQ的分布式事务消息的死信队列消息转移 + 人工处理
文章图片
如果死信队列报错就进行人工处理
文章图片
总结 基于MQ的分布式事务解决方案优点：1、通用性强
2、拓展方便
3、耦合度低，方案也比较成熟
基于MQ的分布式事务解决方案缺点：1、基于消息中间件，只适合异步场景
2、消息会延迟处理，需要业务上能够容忍
建议1、尽量去避免分布式事务
2、尽量将非核心业务做成异步
Springboot整合rabbitmq集群配置详解 1 引入starter
<
parent>
<
groupId>
org.springframework.boot<
/groupId>
<
artifactId>
spring-boot-starter-parent<
/artifactId>
<
version>
2.2.6.RELEASE<
/version>
<
relativePath/>
<
!-- lookup parent from repository -->
<
/parent>
<
dependency>
<
groupId>
org.springframework.boot<
/groupId>
<
artifactId>
spring-boot-starter-amqp<
/artifactId>
<
/dependency>
<
2：详细配置如下
rabbitmq:
addresses: 127.0.0.1:6605,127.0.0.1:6606,127.0.0.1:6705 #指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host)
#port:
##集群配置 addresses之间用逗号隔开
# addresses: ip:port,ip:port
password: admin
username: 123456
virtual-host: / # 连接到rabbitMQ的vhost
requested-heartbeat: #指定心跳超时，单位秒，0为不指定；默认60s
publisher-confirms: #是否启用 发布确认
publisher-reurns: # 是否启用发布返回
connection-timeout: #连接超时，单位毫秒，0表示无穷大，不超时
cache:
channel.size: # 缓存中保持的channel数量
channel.checkout-timeout: # 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel
connection.size: # 缓存的连接数，只有是CONNECTION模式时生效
connection.mode: # 连接工厂缓存模式：CHANNEL 和 CONNECTION
listener:
simple.auto-startup: # 是否启动时自动启动容器
simple.acknowledge-mode: # 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto
simple.concurrency: # 最小的消费者数量
simple.max-concurrency: # 最大的消费者数量
simple.prefetch: # 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量.
simple.transaction-size: # 指定一个事务处理的消息数量，最好是小于等于prefetch的数量.
simple.default-requeue-rejected: # 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
simple.idle-event-interval: # 多少长时间发布空闲容器时间，单位毫秒
simple.retry.enabled: # 监听重试是否可用
simple.retry.max-attempts: # 最大重试次数
simple.retry.initial-interval: # 第一次和第二次尝试发布或传递消息之间的间隔
simple.retry.multiplier: # 应用于上一重试间隔的乘数
simple.retry.max-interval: # 最大重试时间间隔
simple.retry.stateless: # 重试是有状态or无状态
template:
mandatory: # 启用强制信息；默认false
receive-timeout: # receive() 操作的超时时间
reply-timeout: # sendAndReceive() 操作的超时时间
retry.enabled: # 发送重试是否可用
retry.max-attempts: # 最大重试次数
retry.initial-interval: # 第一次和第二次尝试发布或传递消息之间的间隔
retry.multiplier: # 应用于上一重试间隔的乘数
retry.max-interval: #最大重试时间间隔
对于发送方而言，需要做以下配置：1 配置CachingConnectionFactory
2 配置Exchange/Queue/Binding
3 配置RabbitAdmin创建上一步的Exchange/Queue/Binding
4 配置RabbitTemplate用于发送消息，RabbitTemplate通过CachingConnectionFactory获取到Connection，然后想指定Exchange发送
对于消费方而言，需要做以下配置：1 配置CachingConnectionFactory
2 配置Exchange/Queue/Binding
3 配置RabbitAdmin创建上一步的Exchange/Queue/Binding
4 配置RabbitListenerContainerFactory
5 配置 [@RabbitListener] / [@RabbitHandler] 用于接收消息
3 Spring AMQP的主要对象注：如果不了解AMQP请前往官网了解.
文章图片
4 使用：通过配置类加载的方式：
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitConfig {
private static final Logger logger = LoggerFactory.getLogger(RabbitConfig.class);
public static final String RECEIVEDLXEXCHANGE="spring-ex";
public static final String RECEIVEDLXQUEUE="spring-qu1";
public static final String RECEIVEDLXROUTINGKEY="aa";
public static final String DIRECTEXCHANGE="spring-ex";
public static final String MDMQUEUE="mdmQueue";
public static final String TOPICEXCHANGE="spring-top";
@Value("${spring.rabbitmq.addresses}")
private String hosts;
@Value("${spring.rabbitmq.username}")
private String userName;
@Value("${spring.rabbitmq.password}")
private String password;
@Value("${spring.rabbitmq.virtual-host}")
private String virtualHost;
/*@Value("${rabbit.channelCacheSize}")
private int channelCacheSize;
*/
//@Value("${rabbit.port}")
//private int port;
/*@Autowired
private ConfirmCallBackListener confirmCallBackListener;
@Autowired
private ReturnCallBackListener returnCallBackListener;
*/
@Bean
public ConnectionFactory connectionFactory(){
CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
cachingConnectionFactory.setAddresses(hosts);
cachingConnectionFactory.setUsername(userName);
cachingConnectionFactory.setPassword(password);
//cachingConnectionFactory.setChannelCacheSize(channelCacheSize);
//cachingConnectionFactory.setPort(port);
cachingConnectionFactory.setVirtualHost(virtualHost);
//设置连接工厂缓存模式：
cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
//缓存连接数
cachingConnectionFactory.setConnectionCacheSize(3);
//设置连接限制
cachingConnectionFactory.setConnectionLimit(6);
logger.info("连接工厂设置完成，连接地址{}"+hosts);
logger.info("连接工厂设置完成，连接用户{}"+userName);
return cachingConnectionFactory;
}
@Bean
public RabbitAdmin rabbitAdmin(){
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
rabbitAdmin.setAutoStartup(true);
rabbitAdmin.setIgnoreDeclarationExceptions(true);
rabbitAdmin.declareBinding(bindingMdmQueue());
//声明topic交换器
rabbitAdmin.declareExchange(directExchange());
logger.info("管理员设置完成");
return rabbitAdmin;
}
@Bean
public RabbitListenerContainerFactory listenerContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory());
factory.setMessageConverter(new Jackson2JsonMessageConverter());
//最小消费者数量
factory.setConcurrentConsumers(10);
//最大消费者数量
factory.setMaxConcurrentConsumers(10);
//一个请求最大处理的消息数量
factory.setPrefetchCount(10);
//
factory.setChannelTransacted(true);
//默认不排队
factory.setDefaultRequeueRejected(true);
//手动确认接收到了消息
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
logger.info("监听者设置完成");
return factory;
}
@Bean
public DirectExchange directExchange(){
return new DirectExchange(DIRECTEXCHANGE,true,false);
}
@Bean
public Queue mdmQueue(){
Map arguments = new HashMap<
>
();
// 绑定该队列到私信交换机
arguments.put("x-dead-letter-exchange",RECEIVEDLXEXCHANGE);
arguments.put("x-dead-letter-routing-key",RECEIVEDLXROUTINGKEY);
logger.info("队列交换机绑定完成");
return new Queue(RECEIVEDLXQUEUE,true,false,false,arguments);
}
@Bean
Binding bindingMdmQueue() {
return BindingBuilder.bind(mdmQueue()).to(directExchange()).with("");
}
@Bean
public RabbitTemplate rabbitTemplate(){
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
rabbitTemplate.setMandatory(true);
//发布确认
//rabbitTemplate.setConfirmCallback(confirmCallBackListener);
// 启用发布返回
//rabbitTemplate.setReturnCallback(returnCallBackListener);
logger.info("连接模板设置完成");
return rabbitTemplate;
}
/*@Bean
public TopicExchange topicExchange(){
return new TopicExchange(TOPICEXCHANGE,true,false);
}*/
/*
*//**
* @return DirectExchange
*//*
@Bean
public DirectExchange dlxExchange() {
return new DirectExchange(RECEIVEDLXEXCHANGE,true,false);
}
*//*
*
* @return Queue
*//*
@Bean
public Queue dlxQueue() {
return new Queue(RECEIVEDLXQUEUE,true);
}
*//*
* @return Binding
*//*
@Bean
public Binding binding() {
return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(RECEIVEDLXROUTINGKEY);
}*/
}
通过两种方式加载1 通过配置文件
2 通过配置类
说明：上面是通过配置文件与配置类的方式去加载，常用的配置如上所示。
实际使用中要生产方与消费方要分开配置，相关配置也会有小变动，大体配置不变。更多信息可查看官网配置。
集群监控在广大的互联网行业中RabbitMQ几乎都会有集群,那么对于集群的监控就成了企业生态中必不可少的一环。
接下来我们来将讲解主要的4种监控。
管理界面监控管理界面监控需要我们开启对应的插件(rabbitmq-plugins enable rabbitmq_management)
然后访问 [http://ip:15672]
tracing日志监控以下是trace的相关命令和使用（要使用需要先rabbitmq启用插件，再打开开关才能使用）：
| 命令集 |
描述 |
| rabbitmq-plugins list |
查看插件列表 |
| rabbitmq-plugins enable rabbitmq_tracing |
rabbitmq启用trace插件 |
| rabbitmqctl trace_on |
打开trace的开关 |
| rabbitmqctl trace_on -p itcast |
打开trace的开关(itcast为需要日志追踪的vhost) |
| rabbitmqctl trace_off |
关闭trace的开关 |
| rabbitmq-plugins disable rabbitmq_tracing |
rabbitmq关闭Trace插件 |
| rabbitmqctl set_user_tags heima administrator |
只有administrator的角色才能查看日志界面 |
日志追踪rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息--01。");
文章图片
定制自己的监控系统RabbitMQ提供了很丰富的restful风格的api接口，
我们可以通过这些接口得到对应的集群数据，此时我们就可以定制我们的监控系统。
更多API的相关信息和描述可以访问http://ip:15672/api/
接下来我们使用RabbitMQ Http API接口来获取集群监控数据
HttpClient以及Jackson的相关Jar
<
dependency>
<
groupId>
org.apache.httpcomponents<
/groupId>
<
artifactId>
httpclient<
/artifactId>
<
version>
4.3.6<
/version>
<
/dependency>
<
dependency>
<
groupId>
com.fasterxml.jackson.core<
/groupId>
<
artifactId>
jackson-databind<
/artifactId>
<
version>
2.7.4<
/version>
<
/dependency>
<
dependency>
<
groupId>
com.fasterxml.jackson.core<
/groupId>
<
artifactId>
jackson-annotations<
/artifactId>
<
version>
2.7.4<
/version>
<
/dependency>
<
dependency>
<
groupId>
com.fasterxml.jackson.core<
/groupId>
<
artifactId>
jackson-core<
/artifactId>
<
version>
2.7.4<
/version>
<
/dependency>
创建MonitorRabbitMQ类实现具体的代码
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.http.HttpEntity;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.auth.BasicScheme;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
/**
* RabbitMQ的监控
*/
public class MonitorRabbitMQ {
//RabbitMQ的HTTP API——获取集群各个实例的状态信息，ip替换为自己部署相应实例的
private static String RABBIT_NODES_STATUS_REST_URL = "http://192.168.13.111:15672/api/nodes";
//RabbitMQ的HTTP API——获取集群用户信息，ip替换为自己部署相应实例的
private static String RABBIT_USERS_REST_URL = "http://192.168.13.111:15672/api/users";
//rabbitmq的用户名
private static String RABBIT_USER_NAME = "guest";
//rabbitmq的密码
private static String RABBIT_USER_PWD = "guest";
public static void main(String[] args) {
try {
//step1.获取rabbitmq集群各个节点实例的状态信息
Map<
String, ClusterStatus>
clusterMap =
fetchRabbtMQClusterStatus(RABBIT_NODES_STATUS_REST_URL, RABBIT_USER_NAME, RABBIT_USER_PWD);
//step2.打印输出各个节点实例的状态信息
for (Map.Entry entry : clusterMap.entrySet()) {
System.out.println(entry.getKey() + " : " + entry.getValue());
}
//step3.获取rabbitmq集群用户信息
Map<
String, User>
userMap =
fetchRabbtMQUsers(RABBIT_USERS_REST_URL, RABBIT_USER_NAME, RABBIT_USER_PWD);
//step4.打印输出rabbitmq集群用户信息
for (Map.Entry entry : userMap.entrySet()) {
System.out.println(entry.getKey() + " : " + entry.getValue());
}
} catch (IOException e) {
e.printStackTrace();
}
}
public static Map<
String, ClusterStatus>
fetchRabbtMQClusterStatus(String url, String username, String password) throws IOException {
Map<
String, ClusterStatus>
clusterStatusMap = new HashMap<
String, ClusterStatus>
();
String nodeData = https://www.songbingjia.com/android/getData(url, username, password);
JsonNode jsonNode = null;
try {
jsonNode = JsonUtil.toJsonNode(nodeData);
} catch (IOException e) {
e.printStackTrace();
}
Iterator<
JsonNode>
iterator = jsonNode.iterator();
while (iterator.hasNext()) {
JsonNode next = iterator.next();
ClusterStatus status = new ClusterStatus();
status.setDiskFree(next.get("disk_free").asLong());
status.setFdUsed(next.get("fd_used").asLong());
status.setMemoryUsed(next.get("mem_used").asLong());
status.setProcUsed(next.get("proc_used").asLong());
status.setSocketUsed(next.get("sockets_used").asLong());
clusterStatusMap.put(next.get("name").asText(), status);
}
return clusterStatusMap;
}
public static Map<
String, User>
fetchRabbtMQUsers(String url, String username, String password) throws IOException {
Map<
String, User>
userMap = new HashMap<
String, User>
();
String nodeData = https://www.songbingjia.com/android/getData(url, username, password);
JsonNode jsonNode = null;
try {
jsonNode = JsonUtil.toJsonNode(nodeData);
} catch (IOException e) {
e.printStackTrace();
}
Iterator<
JsonNode>
iterator = jsonNode.iterator();
while (iterator.hasNext()) {
JsonNode next = iterator.next();
User user = new User();
user.setName(next.get("name").asText());
user.setTags(next.get("tags").asText());
userMap.put(next.get("name").asText(), user);
}
return userMap;
}
public static String getData(String url, String username, String password) throws IOException {
CloseableHttpClient httpClient = HttpClients.createDefault();
UsernamePasswordCredentials creds = new UsernamePasswordCredentials(username, password);
HttpGet httpGet = new HttpGet(url);
httpGet.addHeader(BasicScheme.authenticate(creds, "UTF-8", false));
httpGet.setHeader("Content-Type", "application/json");
CloseableHttpResponse response = httpClient.execute(httpGet);
try {
if (response.getStatusLine().getStatusCode() != 200) {
System.out.println("call http api to get rabbitmq data return code: " + response.getStatusLine().getStatusCode() + ", url: " + url);
}
HttpEntity entity = response.getEntity();
if (entity != null) {
return EntityUtils.toString(entity);
}
} finally {
response.close();
}
return null;
}
public static class JsonUtil {
private static ObjectMapper objectMapper = new ObjectMapper();
static {
objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
//objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
}
public static JsonNode toJsonNode(String jsonString) throws IOException {
return objectMapper.readTree(jsonString);
}
}
public static class User {
private String name;
private String tags;
@Override
public String toString() {
return "User{" +
"name=" + name +
", tags=" + tags +
\'}\';
}
//GET/SET方法省略
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getTags() {
return tags;
}
public void setTags(String tags) {
this.tags = tags;
}
}
public static class ClusterStatus {
private long diskFree;
private long diskLimit;
private long fdUsed;
private long fdTotal;
private long socketUsed;
private long socketTotal;
private long memoryUsed;
private long memoryLimit;
private long procUsed;
private long procTotal;
// 此处省略了Getter和Setter方法
public long getDiskFree() {
return diskFree;
}
public void setDiskFree(long diskFree) {
this.diskFree = diskFree;
}
public long getDiskLimit() {
return diskLimit;
}
public void setDiskLimit(long diskLimit) {
this.diskLimit = diskLimit;
}
public long getFdUsed() {
return fdUsed;
}
public void setFdUsed(long fdUsed) {
this.fdUsed = fdUsed;
}
public long getFdTotal() {
return fdTotal;
}
public void setFdTotal(long fdTotal) {
this.fdTotal = fdTotal;
}
public long getSocketUsed() {
return socketUsed;
}
public void setSocketUsed(long socketUsed) {
this.socketUsed = socketUsed;
}
public long getSocketTotal() {
return socketTotal;
}
public void setSocketTotal(long socketTotal) {
this.socketTotal = socketTotal;
}
public long getMemoryUsed() {
return memoryUsed;
}
public void setMemoryUsed(long memoryUsed) {
this.memoryUsed = memoryUsed;
}
public long getMemoryLimit() {
return memoryLimit;
}
public void setMemoryLimit(long memoryLimit) {
this.memoryLimit = memoryLimit;
}
public long getProcUsed() {
return procUsed;
}
public void setProcUsed(long procUsed) {
this.procUsed = procUsed;
}
public long getProcTotal() {
return procTotal;
}
public void setProcTotal(long procTotal) {
this.procTotal = procTotal;
}
@Override
public String toString() {
return "ClusterStatus{" +
"diskFree=" + diskFree +
", diskLimit=" + diskLimit +
", fdUsed=" + fdUsed +
", fdTotal=" + fdTotal +
", socketUsed=" + socketUsed +
", socketTotal=" + socketTotal +
", memoryUsed=" + memoryUsed +
", memoryLimit=" + memoryLimit +
", procUsed=" + procUsed +
", procTotal=" + procTotal +
\'}\';
}
}
}
Zabbix 监控RabbitMQZabbix是一个基于WEB界面提供分布式系统监视以及网络监视功能的企业级开源解决方案，
他也可以帮助我们搭建一个MQ集群的监控系统,同时提供预警等功能，
但是由于其搭建配置要求比较高一般都是由运维人员负责搭建，
感兴趣的可以访问https://www.zabbix.com/ 官网进行了解学习。
面试题分析 1、Rabbitmq 为什么需要信道，为什么不是TCP直接通信2：queue到底在消费者创建还是生产者创建？
文章图片
【RabbitMQ-进阶】
推荐阅读
- 学习Linux tar 命令（最简单也最困难）
- 详解逃逸分析标量替换栈上分配
- 数仓中指标-标签，维度-度量，自然键-代理键等各名词解析及关系
- 红帽Linux入门指南第四期（系统的基本操作命令）
- Redis技术探索让你彻底会使用“Redis中最陌生且最强大的集合”（ZSET）前篇
- 最终一致性和实时一致性是什么（在架构设计中，我们应该选择哪种方式？）
- Oracle在MySQL市场掀起了一股热潮
- 一文搞懂jsBridge的运行机制
- 博客总结（导航）