简介
RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue )的开源实现。AMQP 的出现其实也是应了广大人民群众的需求,虽然在同步消息通讯的世界里有很多公开标准(如 COBAR的 IIOP ,或者是 SOAP 等),但是在异步消息处理中却不是这样,只有大企业有一些商业实现(如微软的 MSMQ ,IBM 的 Websphere MQ 等),因此,在 2006 年的 6 月,Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。
RabbitMQ是由RabbitMQ Technologies Ltd开发并且提供商业支持的。该公司在2010年4月被SpringSource(VMWare的一个部门)收购。在2013年5月被并入Pivotal。其实VMWare,Pivotal和EMC本质上是一家的。不同的是VMWare是独立上市子公司,而Pivotal是整合了EMC的某些资源,现在并没有上市。
RabbitMQ的官网是http://www.rabbitmq.com
百度百科amqp协议介绍https://baike.baidu.com/item/AMQP/8354716?fr=aladdin
注意:RabbitMQ是采用erlang语言开发的,所以必须有erlang环境才可以运行
几种常见的消息中间件对比
Erlang语言简介:
Erlang编程语言最初目的是进行大型电信交换设备的软件开发,是一种适用于大规模并行处理环境的高可靠性编程语言。随着多核处理器技术的日渐普及,以及互联网、云计算等技术的发展,该语言的应用范围也有逐渐扩大之势。
百度百科介绍:https://baike.baidu.com/item/Erlang%E8%AF%AD%E8%A8%80/20864044?fr=aladdin
初衷理念实现抗高并发语言
RabbitMQ环境安装
1.下载并安装erlang,下载地址:http://www.erlang.org/download
2.配置erlang环境变量信息
新增环境变量ERLANG_HOME=erlang的安装地址
将%ERLANG_HOME%\bin加入到path中
3.下载并安装RabbitMQ,下载地址:http://www.rabbitmq.com/download.html
注意: RabbitMQ 它依赖于Erlang,需要先安装Erlang。
RabbitMQ管理平台
RabbitMQ 管理平台地址http://127.0.0.1:15672
默认账号:guest/guest 用户可以自己创建新的账号
Virtual Hosts:
像mysql有数据库的概念并且可以指定用户对库和表等操作的权限。那RabbitMQ呢?RabbitMQ也有类似的权限管理。在RabbitMQ中可以虚拟消息服务器VirtualHost,每个VirtualHost相当月一个相对独立的RabbitMQ服务器,每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。
RabbitMQ的5种队列形式
1.点对点(简单)的队列
2.工作(公平性)队列模式
3.发布订阅模式(fanout)
4.路由模式Routing
5.通配符模式Topics
详细网址
http://www.rabbitmq.com/getstarted.html
**
RabbitMQ关键名词
AMQP(高级消息队列协议): 是一个异步消息传递所使用应用层协议规范,为面向消息中间件设计,基于此协议的客户端与消息中间件可以无视消息来源传递消息,不受客户端、消息中间件、不同的开发语言环境等条件的限制;
涉及概念解释:
Server(Broker):接收客户端连接,实现AMQP协议的消息队列和路由功能的进程;
Virtual Host:虚拟主机的概念,类似权限控制组,一个Virtual Host里可以有多个Exchange和Queue。
Exchange:交换机,接收生产者发送的消息,并根据Routing Key将消息路由到服务器中的队列Queue。
ExchangeType:交换机类型决定了路由消息行为,RabbitMQ中有三种类型Exchange,分别是fanout、direct、topic;
Message Queue:消息队列,用于存储还未被消费者消费的消息;
Message:由Header和body组成,Header是由生产者添加的各种属性的集合,包括Message是否被持久化、优先级是多少、由哪个Message Queue接收等;body是真正需要发送的数据内容;
BindingKey:绑定关键字,将一个特定的Exchange和一个特定的Queue绑定起来。
RabbitMQ交换机的作用
生产者发送消息不会向传统方式直接将消息投递到队列中,而是先将消息投递到交换机中,在由交换机转发到具体的队列,队列在将消息以推送或者拉取方式给消费者进行消费,这和我们之前学习Nginx有点类似。
交换机的作用根据具体的路由策略分发到不同的队列中,交换机有四种类型。
Direct exchange(直连交换机)是根据消息携带的路由键(routing key)将消息投递给对应队列的
Fanout exchange(扇型交换机)将消息路由给绑定到它身上的所有队列
Topic exchange(主题交换机)队列通过路由键绑定到交换机上,然后,交换机根据消息里的路由值,将消息路由给一个或多个绑定队列
Headers exchange(头交换机)类似主题交换机,但是头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。
**
消息队列RabbitMQ应答模式
**
为了确保消息不会丢失,RabbitMQ支持消息应答。消费者发送一个消息应答,告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。
如果一个消费者挂掉却没有发送应答,RabbitMQ会理解为这个消息没有处理完全,然后交给另一个消费者去重新处理。这样,你就可以确认即使消费者偶尔挂掉也不会丢失任何消息了。
没有任何消息超时限制;只有当消费者挂掉时,RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。
消息应答是默认打开的。我们通过显示的设置autoAsk=true关闭这种机制。现即自动应答开,一旦我们完成任务,消费者会自动发送应答。通知RabbitMQ消息已被处理,可以从内存删除。如果消费者因宕机或链接失败等原因没有发送ACK(不同于ActiveMQ,在RabbitMQ里,消息没有过期的概念),则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。
RabbitMQ的公平转发
目前消息转发机制是平均分配,这样就会出现俩个消费者,奇数的任务很耗时,偶数的任何工作量很小,造成的原因就是近当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。仅仅盲目转发所有的奇数给一个消费者,偶数给另一个消费者。
为了解决这样的问题,我们可以使用basicQos方法,传递参数为prefetchCount= 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。
换句话说,只有在消费者空闲的时候会发送下一条信息。调度分发消息的方式,也就是告诉RabbitMQ每次只给消费者处理一条消息,也就是等待消费者处理完毕并自己对刚刚处理的消息进行确认之后,才发送下一条消息,防止消费者太过于忙碌,也防止它太过去清闲。
通过 设置channel.basicQos(1);
RabbitMQ的应答模式
案例:
生产者端代码不变,消费者端代码这部分就是用于开启手动应答模式的。
channel.basicConsume(QUEUE_NAME, false, defaultConsumer);
注:第二个参数值为false代表关闭RabbitMQ的自动应答机制,改为手动应答。
在处理完消息时,返回应答状态,true表示为自动应答模式。
channel.basicAck(envelope.getDeliveryTag(), false);
发布/订阅模式Publish/Subscribe
这个可能是消息队列中最重要的队列了,其他的都是在它的基础上进行了扩展。
功能实现:一个生产者发送消息,多个消费者获取消息(同样的消息),包括一个生产者,一个交换机,多个队列,多个消费者。
思路解读(重点理解):
(1)一个生产者,多个消费者
(2)每一个消费者都有自己的一个队列
(3)生产者没有直接发消息到队列中,而是发送到交换机
(4)每个消费者的队列都绑定到交换机上
(5)消息通过交换机到达每个消费者的队列
该模式就是Fanout Exchange(扇型交换机)将消息路由给绑定到它身上的所有队列
以用户发邮件案例讲解
注意:交换机没有存储消息功能,如果消息发送到没有绑定消费队列的交换机,消息则丢失。
路由模式RoutingKey
生产者发送消息到交换机并指定一个路由key,消费者队列绑定到交换机时要制定路由key(key匹配就能接受消息,key不匹配就不能接受消息)
例如:我们可以把路由key设置为insert ,那么消费者队列key指定包含insert才可以接收消息,消费者队列key定义为update或者delete就不能接收消息。很好的控制了更新,插入和删除的操作。
采用交换机direct模式
通配符模式Topics
说明:此模式实在路由key模式的基础上,使用了通配符来管理消费者接收消息。生产者P发送消息到交换机X,type=topic,交换机根据绑定队列的routing key的值进行通配符匹配;
符号#:匹配一个或者多个词lazy.# 可以匹配lazy.irs或者lazy.irs.cor
符号:只能匹配一个词lazy. 可以匹配lazy.irs或者lazy.cor
发布/订阅模式Publish/Subscribe
RabbitMQ消息确认机制
问题产生背景:
生产者发送消息出去之后,不知道到底有没有发送到RabbitMQ服务器, 默认是不知道的。而且有的时候我们在发送消息之后,后面的逻辑出问题了,我们不想要发送之前的消息了,需要撤回该怎么做。
解决方案:
1.AMQP 事务机制
2.Confirm 模式
事务模式:
txSelect 将当前channel设置为transaction模式
txCommit 提交当前事务
txRollback 事务回滚
RabbitMQ消息重试机制
消费者在消费消息的时候,如果消费者业务逻辑出现程序异常,这时候应该如何处理?
答案:使用消息重试机制。(演示重试机制)
如何合适选择重试机制:
情况1: 消费者获取到消息后,调用第三方接口,但接口暂时无法访问,是否需要重试? (需要重试机制)
情况2: 消费者获取到消息后,抛出数据转换异常,是否需要重试?(不需要重试机制)需要发布进行解决。
如何实现重试机制
总结:对于情况2,如果消费者代码抛出异常是需要发布新版本才能解决的问题,那么不需要重试,重试也无济于事。应该采用日志记录+定时任务job健康检查+人工进行补偿
消费者如果保证消息幂等性,不被重复消费
产生原因:网络延迟传输中,消费出现异常或者是消费延迟消费,会造成MQ进行重试补偿,在重试过程中,可能会造成重复消费。
消费者如何保证消息幂等性,不被重复消费
解决办法:
①使用全局MessageID判断消费方使用同一个,解决幂等性。
②或者使用业务逻辑保证唯一(比如订单号码)
RabbitMQ死信队列
死信队列 听上去像 消息“死”了 ,其实也有点这个意思,
死信队列 是 当消息在一个队列 因为下列原因:
1.消息被拒绝(basic.reject或basic.nack)并且requeue=false.
2.消息TTL过期
3.队列达到最大长度(队列满了,无法再添加数据到mq中)
应用场景分析
在定义业务队列的时候,可以考虑指定一个死信交换机,并绑定一个死信队列,当消息变成死信时,该消息就会被发送到该死信队列上,这样就方便我们查看消息失败的原因了
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); 丢弃消息
RabbitMQ解决分布式事务思路
案例:
经典案例,以目前流行点外卖的案例,用户下单后,调用订单服务,让后订单服务调用派单系统通知送外卖人员送单,这时候订单系统与派单系统采用MQ异步通讯。
RabbitMQ解决分布式事务原理: 采用最终一致性原理。
需要保证以下三要素
1、确认生产者一定要将数据投递到MQ服务器中(采用MQ消息确认机制)
2、MQ消费者消息能够正确消费消息,采用手动ACK模式(注意重试幂等性问题)
3、如何保证第一个事务先执行,采用补偿机制,在创建一个补单消费者进行监听,如果订单没有创建成功,进行补单。
docker实例化rabbitmq
docker run -d —name rabbitmq -p 5672:5672 -p 15672:15672 —hostname rabbit -e RABBITMQ_DEFAULT_VHOST=mt-order -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin 985adbf13062
发布/订阅模式Publish/Subscribe
消息队列RabbitMQ应答消息队列RabbitMQ应答模式
模式
式队列
RabbitMQ管理平台
RabbitMQ环境安装
RabbitMQ环境安装
RabbitMQ环境安装RabbitMQ环境安装
RabbitMQ环境安装