RabbiteMq的一些高级特性

本问整理笔记：来自b站视频https://www.bilibili.com/video/BV15k4y1k7Ep?spm_id_from=333.999.0.0

一、RabbiteMq的优势和劣势

优势：

应用解耦

异步提速

削峰填谷

劣势：

系统可用性降低

系统复杂度提高

一致性的问题

二、交换机的机种类型说明

direct 路由指定

fanout 广播类型

topic 主题订阅

三、RabbiteMq高级特性

消息可靠性的投递

ConsumerAck

TTL

死信队列

延迟队列

日志与监控

消息可靠性分析与追踪

1.消息的可靠投递

rabbitmq提供了2中模式来控制消息的投递可靠性

confirm 确认模式
return 退回模式

rabbitemq整个消息的投递的路径为

producer—>rabbitmq broker—>echange->queue->consumer

消息从producer到exchange则会返回一个confirmCallback
消息从exchange–>queue投递失败则会返回一个returnCallback

我们将利用这个2个callback控制消息的可开性投递

操作

设置连接时把参数publish-confirms：true‘

设置回调函数setConfirmCallback

对应参数correlationData相关配置信息，
ack交换机是否成功接收到消息，true成功，false代表事变
cause失败的原因

2.退回模式

可以拿到

seMandatory为true

setReturnCallback()设置消息对象message 消息对象，replytext错误码，exchange交换机，routingkey路由键

可以拿到成功被exchange接收，没有被queue接收的值

总结

设置

2.Consumer Ack

含义解释

ack指Acknowledge ，确认表示消费端收到消息后的确认方式

有三种确认方式：

自动确认：acknowledge=“none” 消息收到自动确认不管业务是否执行成功
手动确认：acknowledge=“manual” 消息收到会先处理业务，业务成功手动调用确认。
根据异常类型确认:acknowledge=“auto”

其中自动确认时指当消息一旦被Consumer接收到时，则自动确认收到，并将相应的message从Rabbitmq的消息从缓存中移除，但是实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失，如果设置了手动确认的方式，则需要再业务处理成功后，调用channel.basicAck(),手动签收，如果出现则调用channel,basicNack()方法，让它自动重新发送消息。

操作步骤

1.设置手动签收，acknowledge=“manual”

2.让监听器类实现ChannelLAwareMessagelListener接口

3.如果消息成功处理，则调用channel的basicAck签收

4.如果消息处理失败则调用channel的basicNack拒绝签收

3.consumer限流机制

1.确保ack机制为手动确认

2.listener-cintainer配置属性

perfetch=1 表示消费者每次从mq拉取一条消息来消费，直到手动确认消费完成后，才会继续拉取一条消息。

4.TTL

TTL全称time to live 存活时间/过期时间

当消息到达存活时间后，还没有被消费，会被自动清楚

RabbiteMQ可以消息设置过
期时间，也可以对整个队列设置过去时间。

小结

设置队列过期时间使用参数：x-message-ttl，单位ms（毫秒），会对整个队列消息统一过期。
设置消息过期时间使用参数expiration，单位：ms（毫秒），当该消息再队列头部时（消费时），会单独判断这个消息是否过期
如果2个者都是进行了设置，以时间短的为准

5.死信队列

死信队列，缩写。英文缩写：DLX，Dead Letter Exchange（死信交换机），当消息成为Dead ，message后。可以重新发送给另一个交换机，这个交换机就是DLX.

消息成为死信的三种情况：

1.队列消息长度到达限制

2.消费者拒绝消费消息，basicNack/basicReject,并且不把消息放入原目标队列，requeue=false

3.原队列存在消息过期设置，消息到达超时间未被消费

队列设置死信交换机：

给队列设置x-dead-letter-exchange 和x-dead-letter-routing-key

5.延迟队列

延迟队列，即消息进入队列后不会被立即消费，只有到达指定时间后，才会被消费

需求：

1.下单后，30分钟未支付，取消订单，回滚库存

2.新用户注册成功7天后，发送消息问候

实现方式：

1.定时器

2.延迟队列

但是rabbitemq没有直接延迟队列功能，但是可以通过ttl和死信队列完成延迟功能

6.日志与监控

1.rabbitmqctl管理和监控

rabbitmqctl list_queues 查看队列

rabbitmqctl list_vhost 查看虚拟机

rabbitmqctl environment 查看环境变量

rabbitmqctl list_exchanges 查看交换机

rabbitmqctl list_queues name messages_unacknowledged 查看未被确认的队列

rabbitmqctl list_users 查看用户

rabbitmqctl list_queues name memory 查看单个队列的内存使用

rabbitmqctl list_connections 查看连接

rabbitmqctl list_queues name messages_ready 查看准备就绪的队列

rabbitmqctl list_consumers 查看消费者信息

7.消息的追踪

在使用任何消息中间件的过程中，难免会出现某条消息异常丢失的情况，对于Rabbitmq而言，可能时因为生产和消费者与rabbitmq断开了连接，而他与rabbitmq又采用了不同确认机制，也有可能因为交换机与队列之间不同的转发策略，甚至是交换机并没有与任何队列进行绑定，生产者又不敢吱或者没有采取相应的措施，另外rabbbitmq本省的集群策略也可能导致消息的丢失，这个时候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。

在rabbitmq中可以使用Firehose和rabbitmq_tracing插件功能来实现消息的追踪。

1.消息追踪firehose

打开trace会影响消息写入功能，不建议生产环境打开。

开启插件 rabbitmqctl trace_on 关闭插件 rabbitmqctl trace_ff

2.rabbitmq插件

rabbitmq_tracing和firehose在实现上如出一辙，只不过rabbitmq_tracing方式比Firehose多了一层GUI包装，更容易使用和管理

启用插件：rabbitmq-plugings enable rabbitmq_tracing

8.消息的可靠新保障

需求：确保100%发送成功

消息补偿机制

消息的幂等性保障

幂等性指一次和多次请求某个资源，对于资源本身应该具有相同的结果，也就是说，其认一多次执行资源本身所产生的影响均与一次执行的影响相同。

在MQ中指，消费多条相同消息，得到与消费该消息一次相同的结果。

幂等性的乐观锁机制

9.RabbiteMQ应用消息

消息可靠性的保障

消息的幂等性处理（就是消息的重复性处理）

10.RabbitMQ的集群的搭建

RabbitMq这款消息队列中间件产品本身是级域Erlang编写，Erlang语言天生具备分布式特性（通过Erlang记群各节点magic cookie 来实现）。因此，RabbitMQ天然支持Clustering，这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过zookeeper分别来实现HA方案和保存集群的元数据，集群是保证可靠性的一种方式，同事可以通过水平扩展以达到增加消息吞吐量能力的目的。

单机多实例部署

首先确保mq运行正常 rabbitmqctl status
停止服务 service rabbitmq-server stop
启动第一个节点 RABBITMQ_NODE_PROT=5673 RABBITMQ_NODENAME= rabbit1 rabbitmq-server start
重打开个win窗口，启动第二个接节点 RABBITMQ_NODE_PROT=5674 RABBITMQ_NODENAME= rabbit2 rabbitmq-server start

结束命令 rabbitmqctl -n rabbit1 stop rabbitmqctl -n rabbit2 stop

rabbit1操作为主节点：

rabbitmqctl -n rabbit1 stop_app
rabbitmqctl -n rabbit1 restart
rabbitmqctl -n rabbit1 start_app

rabbit2操作为从节点：

rabbitmqctl -n rabbit2 stop_app
rabbitmqctl -n rabbit2 restart
rabbitmqctl -n rabbit2 join_cluster rabbit1@'super' ###''内是主机名换成自己的
rabbitmqctl -n rabbit2 start_app

查看集群状态

cluster_status -n rabbit1

目前虽然构成了集群但是数据是不同步的

11.Rabbitmq增加镜像队列

如下如创建

12.负载均衡HAProxy

就是解决上面很多节点，地址很多，连接写哪个都不好写，装了这个就可以统一写一个内部协调处理

HAProxy提供高可用、负载均衡以及基于Tcp和http应哟的代理。支持虚拟主机，它是免费、快速并且可靠的一中解决方案，包括witter，reddit，stackOverflow，github在内的多家知名互联网公司在使用，HAProxy实现了一种事件驱动、单一进程模型，此模型支持非常大的并发连接数。

安装

//下载依赖包
yum  install gcc vim wget
//上传haproxy源码包
wget http://www.haproxy.org/download/1.6/src/haproxy-1.6.5.tar.gz
//解压
tar....

配置文件

启动HAProxy负载

/usr/local/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg
//查看haproxy进程状态
ps -ef | grep haproxy
访问如下地址对mq节点进行监控
http://172.16.133:8100/rabbitmq-stats

实际代码里面连接写但是hqproxy的地址加：5672，即172.16.133:5672