rabbitmq实战指南(rabbitmq使用案例)

RabbitMQ学习

1.MQ概述2.MQ的优势与劣势

2.1优势2.2劣势 3.常见的MQ产品4.RabbitMQ简介

4.1工作模式4.2 JMS 5. 快速入门

5.1生产者5.2消费者 6.RabbitMQ的工作模式

6.1简单模式(一对一)6.2工作队列模式(一对多)6.3订阅模式(引入交换机)6.3路由模式6.4通配符模式 7.RabbitMQ高级特性

7.1消息的可靠性投递

7.1.1确认模式7.1.2回退模式 7.2Consumer Ack7.3消费端限流7.4TTL7.5死信队列(死信交换机)7.6延迟队列7.7日志与监控7.8消息追踪 8.RabbitMQ应用问题

8.1消息可靠性保障(消息补偿机制)8.2消息幂等性保障(乐观锁解决方案)

1.MQ概述

MQ全程Message Queue(消息队列),实在消息党的传输过程中保存消息的容器,多用于分布式系统之间进行通信

MQ,消息队列,存储消息的中间件分布式系统通信有两种方式

直接远程调用借助第三方完成间接通信 发送方成为生产者,接收方成为消费者 2.MQ的优势与劣势 2.1优势

应用解耦

提高容错性,提高可维护性 异步提速

提升用户体验,提高系统吞吐量 削峰填谷

提高系统稳定性 2.2劣势

系统可用性降低

系统引入的外部依赖越多,系统稳定性越差,需要保证高可用 系统复杂度提高

以前同步的远程调用,现在MQ异步调用,需要保证消息不被重复消费,处理消息丢失情况,保证消息传递的顺序性 一致性问题

MQ调用的系统可能不是全部成功,需要保证消息数据处理一致性

MQ需要满足的条件

生产者不需要从消费者处获得反馈.引入消息队列之前的直接调用,其接口返回值应该为空,这才可以使异步成为可能允许短暂的不一致性好处要超过使用以及管理成本

3.常见的MQ产品

RabbitMQ

使用Erlang语言开发,社区活跃,管理界面丰富,消息延迟为微秒级(最低),单机吞吐量达到万级 ActiveMQ

使用Java语言开发,老牌产品,成熟度高,文档较多,消息延迟为毫秒级,但是单机吞吐量最差,为万级 RocketMQ

使用Java语言开发,功能较为晚辈,但是扩展性差,消息延迟达到毫秒级,但是社区活跃低,单机吞吐量最高,达到十万级 Kafka

使用Scala&Java语言开发,只支持主要的MQ功能,因为主要为大数据领域准备,消息延迟在毫秒内,单机吞吐量十万级 4.RabbitMQ简介

基于AMQP(高息消息队列协议),是一个网络协议,是应用层协议的一个开放标准,为面向消息的中间件设计.基于此协议的客户端与消息中间线可传递消息,并不受客户端中间件的不同产品、不同开发语言等条件限制.2006年,AMQP规范发布.

4.1工作模式

RabbitMQ提供了6种工作模式

简单模式worl queues工作队列模式Publish/Subscribe发布于订阅模式Routing路由模式Topics主题模式RPC远程调用模式 4.2 JMS

JMS是Java消息服务应用程序接口,是一个Java平台种关于面向消息中间件的APIJMS是JavaEE规范中的一种很多消息中间件都实现了JMS规范,例如ActiveMQ. RabbitMQ官方没有实现包,但是开源社区有 5. 快速入门 5.1生产者

创建连接工厂

ConnectionFactory factory = new ConnectionFactory

设置参数

factory.setHost("xxx.xxx.xxx.xxx");//ip 默认值localhost
factory.setPort(5679);//端口
factory.setVirtualHost("xxx");//虚拟机名
factory.setUsername("xxx");//用户名
factory.setPassword("xxx");//密码

创建链接Connection

Connection connection = factory.newConnection();

创建Channel

Channel channel = connection.create Channel();

创建队列Queue

channel.quueDelare(...);

参数列表

queue 队列名称durable 是否持久化exclusive 是否独占,连接关闭时是否删除队列autoDelete 没有消费者是否自动删除arguments 参数

发送消息

channel.basicPublish(...);

参数列表

exchange 交换机名称,简单模式下使用默认,""routingKey 路由名称props 配置信息body 发送消息数据

释放资源

channel.close();
connection.close();

5.2消费者

创建连接工厂

设置参数

创建连接

创建Channel

创建队列

接收消息

channel.basicConsume(...);

参数列表

queue 队列名称

autoAck 是否自动确认

callback 回调对象(消费者)

Consume consume = new DefaultConsumer(channel){
    @Override
    public void hadleDelivery(....)
}

参数列表

consumerTag标识envelope 获取一些信息(生产者设置的参数)body 接收的数据

消费者不释放资源

6.RabbitMQ的工作模式 6.1简单模式(一对一)

如上

6.2工作队列模式(一对多)

说明

与入门程序的简单模式相比,多了与一个或一些消费端,多个消费端共同消费同一个队列的消息

应用场景

对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度 6.3订阅模式(引入交换机)

多了交换机(简单模式使用默认的)

交换机:只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与交换机绑定,或者没有符合路由规则的队列,消息会丢失

交换机类型

广播,将小交给所有绑定到交换机的队列定向,把消息交给符合指定routing key的队列通配符,把消息交给符合路由模式的队列 生产者

创建连接工厂设置参数创建连接创建Channel创建交换机,指定类型fanout创建队列绑定队列和交换机发送消息释放资源 6.3路由模式

说明

队列与交换机的绑定,不是任意绑定,而是指定RoutingKey消息发送方在向交换机发送消息时也必须指定消息的RoutingKey交换机会根据消息的RoutingKey进行判断,只有队列与消息的RoutingKey一致才会判断

生产者

创建连接工厂设置参数创建连接创建Channel创建交换机,指定类型routing创建队列绑定队列和交换机(指定RoutingKey)发送消息(指定RoutingKey)释放资源 6.4通配符模式

说明

更加灵活*:多个字母#:零个或多个字母 生产者

创建连接工厂设置参数创建连接创建Channel创建交换机,指定类型topic创建队列绑定队列和交换机(指定通配符)发送消息(指定通配符)释放资源 7.RabbitMQ高级特性 7.1消息的可靠性投递

RabbitMQ的整个消息投递路径

producer—>rabbitmq broker—>exchange—queue—>consumer

在使用RabbitMQ的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败的场景,RabbitMQ提供了两种方式来控制消息的投递可靠性模式

confirm 确认模式

消息从producer到exchange则会返回一个/confirm/iCallback return 退回模式

消息从exchange—>queue投递失败则会返回一个returnCallback 7.1.1确认模式

步骤

确认模式开启:ConnectionFactory种开启publisher-confirms="true"

在rebbitTemplate定义/confirm/iCallback回调函数

rebbitTemplate.set/confirm/iCallback(new RabbitTamplate./confirm/iCallback(){
    
    @Override
    public void /confirm/i(CorrelationData correlationData,boolean ack,String cause){
        //...业务代码
    }
});

7.1.2回退模式

步骤

开启回退模式

设置Exchange处理消息的模式

如果消息没有路由到Queue,则丢弃消息(默认)如果消息没有到Queue,返回消息给发送方ReturnCallback

rabbitTempalte.setMandatory(true);//设置为如果失败返回给发送方

设置ReturnCallback publisher-return="true"

rebbitTemplate.setReturnCallback(new RabbitTamplate.ReturnCallback(){
        
    @Override    
    public void returnedMessage(Message message,int replyCode,String replyText,String exchange,String routingKey){
        //...业务代码    
    }
});

7.2Consumer Ack

消费端收到消息后的确认方式

三种确认方式

自动确认 acknowledge="none"

收到就确认

手动确认 acknowledge="manual"

业务处理成功后才手动确认,监听器实现ChannelAwareMessageListener接口

channel.basicAck()成功后调用,手动签收

channel.basicNack()出现异常后调用,使其自动重发

根据异常情况确认 acknowledge="auto"

7.3消费端限流

步骤

确保ack机制为手动确认listener-container配置属性

perfetch = 1,标识消费端每次从mq拉取一条消息来消费,知道手动确认消费完毕,才会继续拉取下一条消息

7.4TTL

TTL全称Time To Live(存活时间/过期时间)

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

RabbitMQ可以对消息设置过期时间,也可以对整个队列(Queue)设置过期时间

两种类型

队列同一过期

设置queue参数,过期时间

(队列过期后,会将队列所有消息全部移除)

消息单独过期

消息后处理对象,设置消息参数

(消息过期后,只有在顶端才会被移除)

都设置了,以短的为准

7.5死信队列(死信交换机)

当消息成为死信后,可以被重新发送到另一个交换机,这个交换机就是DLX(死信交换机)

死信概念

队列长度到达限制消费者拒接消费消息,并且不把消息重新放回原目标队列原队列存在消息过期设置,消息到达超时时间未被消费

队列绑定死信交换机

给队列设置参数

x-dead-letter-exchange 死信交换机名称

x-dead-letter-routing-key 发送私信交换机的routingkey

7.6延迟队列

概念

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

RabbitMQ没有提供延迟队列功能,但是可以使用TTL+死信队列的方式实现延迟队列

延迟队列效果实现,消费者监听的是死信队列

7.7日志与监控

日志

RabbitMQ默认日志存放路径: /var/log/rabbitmq/rabbit@主机名.log

web管控台监控

rabbitmqctl 管理和监控

7.8消息追踪

在RabbitMQ种可以使用Firehose和rabbitmq_tracing插件管理实现消息追踪

Firehose

rabbitmqctl trace_on 开启

rabbitmqctl trace_off 关闭

rabbitmq_tracing

rabbitmq-plugins enable rabbitmq_tracing 开启插件

与firehose实现相同,只不过,rabbitmq_tracing多了一层GUI包装

8.RabbitMQ应用问题 8.1消息可靠性保障(消息补偿机制)

回调检查服务,在正常发送消息后延迟发送相同消息,通过是否确认进行检查是否消费成功

8.2消息幂等性保障(乐观锁解决方案)

幂等性

一次或多次请求某一资源,对于资源本身应该具有同样的结果.(消费多条的消息,与消费该消息一次取得相同的结果)

使用版本号进行保障