淘先锋技术网

首页 1 2 3 4 5 6 7

SpringBoot 整合 RabbitMQ

概念

2007 年发布,是一个在 AMQP(高级消息队列协议)基础上完成的,可复用的企业消息系统,是当前最主流的消息中间件之一。
RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue 高级消息队列协议 )的开源实现,由于erlang 语言的高并发特性,性能较好,本质是个队列,FIFO 先入先出,里面存放的内容是message。
RabbitMQ 是一个消息中间件:它接收消息并且转发,就类似于一个快递站,卖家把快递通过快递站,送到我们的手上,MQ也是这样,接收并存储消息,再转发。

MQ典型应用场景:

  • 异步处理。把消息放入消息中间件中,等到需要的时候再去处理。
  • 流量削峰。例如秒杀活动,在短时间内访问量急剧增加,使用消息队列,当消息队列满了就拒绝响应,跳转到错误页面,这样就可以使得系统不会因为超负载而崩溃。
  • 日志处理
  • 应用解耦。假设某个服务A需要给许多个服务(B、C、D)发送消息,当某个服务(例如B)不需要发送消息了,服务A需要改代码再次部署;当新加入一个服务(服务E)需要服务A的消息的时候,也需要改代码重新部署;另外服务A也要考虑其他服务挂掉,没有收到消息怎么办?要不要重新发送呢?是不是很麻烦,使用MQ发布订阅模式,服务A只生产消息发送到MQ,B、C、D从MQ中读取消息,需要A的消息就订阅,不需要了就取消订阅,服务A不再操心其他的事情,使用这种方式可以降低服务或者系统之间的耦合。

首先先介绍一个简单的一个消息推送到接收的流程,提供一个简单的图:

JCccc-RabbitMq

黄色的圈圈就是我们的消息推送服务,将消息推送到 中间方框里面也就是 rabbitMq的服务器,然后经过服务器里面的交换机、队列等各种关系(后面会详细讲)将数据处理入列后,最终右边的蓝色圈圈消费者获取对应监听的消息。

常用的交换机有以下三种,因为消费者是从队列获取信息的,队列是绑定交换机的(一般),所以对应的消息推送/接收模式也会有以下几种:

Direct Exchange

直连型交换机,根据消息携带的路由键将消息投递给对应队列。

大致流程,有一个队列绑定到一个直连交换机上,同时赋予一个路由键 routing key 。
然后当一个消息携带着路由值为X,这个消息通过生产者发送给交换机时,交换机就会根据这个路由值X去寻找绑定值也是X的队列。

Fanout Exchange

广播式交换机,这个交换机没有路由键概念,就算你绑了路由键也是无视的。 这个交换机在接收到消息后,会直接转发到绑定到它上面的所有队列。

Topic Exchange

主题交换机,这个交换机其实跟直连交换机流程差不多,但是它的特点就是在它的路由键和绑定键之间是有规则的。

其路由规则为:routingkey必须为单词列表,单词之间以点号分隔开,*号代表一个单词,#号可以替代零个或多个单词。

举例:

路由*.*.key1绑定队列Q1,如:red.blue.key1、black.green.key1等,通过这些路由都可以发送消息给队列Q1

路由#.key2.*绑定队列Q2,如:a.b.c.key2.red、orange.key2.blue、key2.red等,通过这些路由都可以发送消息给队列Q2

路由key3.#绑定队列Q2,如:key3.red.blue、key3、key3.green、key3.black.blue.red等,通过这些路由都可以将消息发送给Q2

主流的消息队列对比

特性ActiveMQRabbitMQRocketMQKafka
单机吞吐量万级,比 RocketMQ、Kafka 低一个数量级同 ActiveMQ10 万级,支撑高吞吐10 万级,高吞吐,一般配合大数据类的系统来进行实时数据计算、日志采集等场景
topic 数量对吞吐量的影响topic 可以达到几百/几千的级别,吞吐量会有较小幅度的下降,这是 RocketMQ 的一大优势,在同等机器下,可以支撑大量的 topictopic 从几十到几百个时候,吞吐量会大幅度下降,在同等机器下,Kafka 尽量保证 topic 数量不要过多,如果要支撑大规模的 topic,需要增加更多的机器资源
时效性ms 级微秒级,这是 RabbitMQ 的一大特点,延迟最低ms 级延迟在 ms 级以内
可用性高,基于主从架构实现高可用同 ActiveMQ非常高,分布式架构非常高,分布式,一个数据多个副本,少数机器宕机,不会丢失数据,不会导致不可用
消息可靠性有较低的概率丢失数据基本不丢经过参数优化配置,可以做到 0 丢失同 RocketMQ
功能支持MQ 领域的功能极其完备基于 erlang 开发,并发能力很强,性能极好,延时很低MQ 功能较为完善,还是分布式的,扩展性好功能较为简单,主要支持简单的 MQ 功能,在大数据领域的实时计算以及日志采集被大规模使用

SpringBoot整合RabbitMQ

​ RabbitMQ是MQ产品中的目前较为流行的产品之一,它遵从AMQP协议。RabbitMQ的底层实现语言使用的是Erlang,所以安装RabbitMQ需要先安装Erlang。

Erlang安装

​ windows版安装包下载地址:https😕/www.erlang.org/downloads

​ 下载完毕后得到exe安装文件,一键傻瓜式安装,安装完毕需要重启,需要重启,需要重启。

​ 安装的过程中可能会出现依赖Windows组件的提示,根据提示下载安装即可,都是自动执行的,如下:

​ Erlang安装后需要配置环境变量,否则RabbitMQ将无法找到安装的Erlang。需要配置项如下,作用等同JDK配置环境变量的作用。

  • ERLANG_HOME
  • PATH

安装 rabbitMQ

​ csdn 安装教程:https://blog.csdn.net/tirster/article/details/121938987

​ windows版安装包下载地址:https://rabbitmq.com/install-windows.html

​ 下载完毕后得到exe安装文件,一键傻瓜式安装,安装完毕后会得到如下文件

启动服务器

rabbitmq-service.bat start		# 启动服务
rabbitmq-service.bat stop		# 停止服务
rabbitmqctl status				# 查看服务状态

​ 运行sbin目录下的rabbitmq-service.bat命令即可,start参数表示启动,stop参数表示退出,默认对外服务端口15672。

​ 注意:启动rabbitmq的过程实际上是开启rabbitmq对应的系统服务,需要管理员权限方可执行。

​ 说明:有没有感觉15672的服务端口很熟悉?activemq与rabbitmq有一个端口冲突问题,学习阶段无论操作哪一个?请确保另一个处于关闭状态。

​ 说明:不喜欢命令行的小伙伴可以使用任务管理器中的服务页,找到RabbitMQ服务,使用鼠标右键菜单控制服务的启停。

访问web管理服务

​ RabbitMQ也提供有web控制台服务,但是此功能是一个插件,需要先启用才可以使用。

rabbitmq-plugins.bat list							# 查看当前所有插件的运行状态
rabbitmq-plugins.bat enable rabbitmq_management		# 启动rabbitmq_management插件

​ 启动插件后可以在插件运行状态中查看是否运行,运行后通过浏览器即可打开服务后台管理界面

http://localhost:15672

​ web管理服务默认端口15672,访问后可以打开RabbitMQ的管理界面,如下:

[图片加载失败]

​ 首先输入访问用户名和密码,初始化用户名和密码相同,均为:guest,成功登录后进入管理后台界面,如下:

[图片加载失败]

整合(direct模型)

​ RabbitMQ满足AMQP协议,因此不同的消息模型对应的制作不同,先使用最简单的direct模型开发。

步骤①:导入springboot整合amqp的starter,amqp协议默认实现为rabbitmq方案

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

步骤②:配置RabbitMQ的服务器地址

spring:
  rabbitmq:
    host: localhost
    port: 5672

步骤③:初始化直连模式系统设置

​ 由于RabbitMQ不同模型要使用不同的交换机,因此需要先初始化RabbitMQ相关的对象,例如队列,交换机等

@Configuration
public class RabbitConfigDirect {
    // 声明队列
    @Bean
    public Queue directQueue(){
        return new Queue("direct_queue");
    }
    @Bean
    public Queue directQueue2(){
        return new Queue("direct_queue2");
    }
    // 声明交换机
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("directExchange");
    }
  	// 声明交换机和队列的绑定关系
    @Bean
    public Binding bindingDirect(){
        return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");
    }
    @Bean
    public Binding bindingDirect2(){
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("direct2");
    }
}

​ 队列Queue与直连交换机DirectExchange创建后,还需要绑定他们之间的关系Binding,这样就可以通过交换机操作对应队列。

步骤④:使用AmqpTemplate操作RabbitMQ

@Service
public class MessageServiceRabbitmqDirectImpl implements MessageService {
    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理队列(rabbitmq direct),id:"+id);
        amqpTemplate.convertAndSend("directExchange","direct",id);
    }
}

​ amqp协议中的操作API接口名称看上去和jms规范的操作API接口很相似,但是传递参数差异很大。

步骤⑤:使用消息监听器在服务器启动后,监听指定位置,当消息出现后,立即消费消息

@Component
public class MessageListener {
    @RabbitListener(queues = "direct_queue")
    public void receive(String id){
        System.out.println("已完成短信发送业务(rabbitmq direct),id:"+id);
    }
}

​ 使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。

整合(topic模型)

步骤①:同上

步骤②:同上

步骤③:初始化主题模式系统设置

@Configuration
public class RabbitConfigTopic {
    @Bean
    public Queue topicQueue(){
        return new Queue("topic_queue");
    }
    @Bean
    public Queue topicQueue2(){
        return new Queue("topic_queue2");
    }
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("topicExchange");
    }
    @Bean
    public Binding bindingTopic(){
        return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.*.id");
    }
    @Bean
    public Binding bindingTopic2(){
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.orders.*");
    }
}

​ 主题模式支持routingKey匹配模式,*表示匹配一个单词,#表示匹配任意内容,这样就可以通过主题交换机将消息分发到不同的队列中,详细内容请参看RabbitMQ系列课程。

匹配键topic.*.*topic.#
topic.order.idtruetrue
order.topic.idfalsefalse
topic.sm.order.idfalsetrue
topic.sm.idfalsetrue
topic.id.ordertruetrue
topic.idfalsetrue
topic.orderfalsetrue

步骤④:使用AmqpTemplate操作RabbitMQ

@Service
public class MessageServiceRabbitmqTopicImpl implements MessageService {
    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void sendMessage(String id) {
        System.out.println("待发送短信的订单已纳入处理队列(rabbitmq topic),id:"+id);
        amqpTemplate.convertAndSend("topicExchange","topic.orders.id",id);
    }
}

​ 发送消息后,根据当前提供的routingKey与绑定交换机时设定的routingKey进行匹配,规则匹配成功消息才会进入到对应的队列中。

步骤⑤:使用消息监听器在服务器启动后,监听指定队列

@Component
public class MessageListener {
    @RabbitListener(queues = "topic_queue")
    public void receive(String id){
        System.out.println("已完成短信发送业务(rabbitmq topic 1),id:"+id);
    }
    @RabbitListener(queues = "topic_queue2")
    public void receive2(String id){
        System.out.println("已完成短信发送业务(rabbitmq topic 22222222),id:"+id);
    }
}

​ 使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。

总结

  1. springboot整合RabbitMQ提供了AmqpTemplate对象作为客户端操作消息队列
  2. 操作ActiveMQ需要配置ActiveMQ服务器地址,默认端口5672
  3. 企业开发时通常使用监听器来处理消息队列中的消息,设置监听器使用注解@RabbitListener
  4. RabbitMQ有5种消息模型,使用的队列相同,但是交换机不同。交换机不同,对应的消息进入的策略也不同