Java开发 - 消息队列之RabbitMQ初体验

Posted CodingFire

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java开发 - 消息队列之RabbitMQ初体验相关的知识,希望对你有一定的参考价值。

目录

前言

RabbitMQ

什么是RabbitMQ

RabbitMQ特点

安装启动

RabbitMQ和Kafka的消息收发区别

RabbitMQ使用案例

添加依赖

添加配置

创建RabbitMQ配置类

RabbitMQ消息的发送

RabbitMQ消息的接收

测试

结语


前言

前一篇,我们学习了Kafka的基本使用,这一篇,我们来学习RabbitMQ。他们作为消息队列本身都具有很强大的功能,博主写完后的感受是:这也只能算是初体验了。毕竟师父领进门,修行靠个人,但是博主怎么会做这种师傅呢?一定要手把手教会各位童鞋,暂时不去深入的讲解了,这一块内容,等博主研究准备好之后,会在后续的系列中对消息队列来一个进阶的教程,大家先入个门,慢慢学习,微服务的东西很多,先学会才是正道。

RabbitMQ

RabbitMQ多用于Java,和Kafka多用于大数据不同,RabbitMQ更偏向于功能,性能一般,所以在Java开发中,比较受欢迎。那么下面,我们就来了解一下RabbitMQ及其特点。

什么是RabbitMQ

RabbitMQ是使用Erlang语言开发的开源消息队列系统,基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。AMQP协议更多用在企业系统内,对数据一致性、稳定性和可靠性要求很高的场景,对性能和吞吐量的要求还在其次。所以在Java业务中使用很多,目前我所知道的很多都是用的此消息队列,只不过大家评价都是:功能强,性能一般。与其相反的,就是Kafka了:性能强,功能弱。

这是因为他们设计的目的不一样,RabbitMQ 在有大量消息堆积的情况下性能会下降,其优势体现在功能上。而Kafka一开始是用来处理海量日志的,所以体现出来就是性能强,功能弱。

所谓,尺有所短,寸有所长,正是这个道理。全都兼顾的几乎是不存在的。

RabbitMQ特点

本想去百科里复制出来,哈哈,结果百科写的太简单了:

  • 可伸缩性:集群服务
  • 消息持久化:从内存持久化消息到硬盘,再从硬盘加载到内存

真是想偷懒都没办法。

RabbitMQ拥有数以万计的用户,是最受欢迎的开源消息代理之一。从T-MobileRuntastic,RabbitMQ在全球范围内用于小型初创企业和大型企业。

RabbitMQ体量轻,易于在本地和云端部署。它支持多种消息协议。RabbitMQ可以部署在分布式和联合配置中,以满足大规模、高可用性的要求。

RabbitMQ可在许多操作系统和云环境中运行,并为大多数流行的语言提供大量的开发工具

官网地址:RabbitMQ Tutorials — RabbitMQ

RabbitMQ的可靠性:体现在持久化、传输确认和发布确认上。

Flexible Routing:灵活路由,消息进入队列之前,通过RabbitMQ内置的Exchange来路由消息 ,针对一些复杂路由,还能将多个Exchange绑定在一起,通过插件实现自己的Exchange。

集群:这个是很多微服务软件都有的功能,比如Redis,ES等,通过此功能,可以实现高可用性的能力,在一个节点发生故障时,其他节点能快速使用。

多协议:RabbitMQ支持多种协议,具体看这里:Which protocols does RabbitMQ support? — RabbitMQ

多语言客户端:RabbitMQ 几乎支持所有常用语言,具体可查看这里:Clients Libraries and Developer Tools — RabbitMQ 

管理监控:这个我们在下面安装的时候会提到此后台页面,还会登录上去给大家看,类似于nacos的管理页面,通过此界面,可以管理和监控RabbitMQ。

插件:RabbitMQ 提供了许多插件,可多方面扩展,也可以自定义插件,详情可看这里:Plugins — RabbitMQ

以上,可通过访问此链接查看详细内容:Messaging that just works — RabbitMQ 

安装启动

这也是本篇内容最痛苦的一个地方,因为RabbitMQ是Erlang语言开发的,所以要先安装Erlang语言的运行环境,为了不去分别讲解Mac和Windows的安装方式,博主准备偷个懒,让大家也偷个懒。

首先,看到这里,足以说明是至少是一名初级Java工程师,你至少开发过一个项目,那么你一定知道Docker,所以,下面,让我们愉快的使用Docker来安装RabbitMQ吧。

在Docker中搜索RabbitMQ,下载rabbitmq:management:

 你下载上面的那个也是可以的,都没关系。

接着打开终端,在里面直接输入,记住,是直接输入,不用进任何目录:

docker run -d --hostname localhost --name rabbitmq -p 15672:15672 -p 5672:5672 rabbitmq:management

 

这个命令输入完成,你的 rabbitmq容器就已经跑起来了:

此时,为了测试rabbitmq是否启动成功,我们需要在浏览器输入:http://localhost:15672

打开后是一个登录页面:

因为没有设置登录的账户名和密码,所以我们采用默认的账户密码guest来登录,输入账户密码都为guest即可登录:

 

如果你想添加用户密码,那就来这里:

 

如果你想修改当前用户密码,你可点击此处:

 

有个Update this user,输入新密码即可: 

 

 

如果你不想使用当前用户名,那就新增一个用户,再删除原来的用户。到这里,安装启动这一环节就结束了,你说,博主是不是偷了个大懒呢? 

RabbitMQ和Kafka的消息收发区别

Kafka使用话题名称来收发信息,同一个话题,一个发,可多个收,使用起来结构简单易懂。

RabbitMQ则要略复杂些,它是通过交换机和路由key来指定要发送的消息队列,生产者发送消息时指定交换机和路由key,这样一个确定的消息队列就产生了,而消费者只需要指定这个产生的消息队列就可以接收到消息。

所以在编写代码时,RabbitMQ会比Kafka多一个配置类,确切的说,是每个业务都要有一个配置类,这个配置类坐的就是指定交换机和路由key,再由路由key指定队列的工作。

RabbitMQ使用案例

此处,我们依然采用前面的微服务项目作为基础,在stock模块来完成RabbitMQ的简单使用。

添加依赖

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

添加配置

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

用户名和密码要用你上面管理页面添加的 有效用户名密码,不添加,默认guest。

创建RabbitMQ配置类

上面提到过,交换机,路由key,队列,这三者的指定需要在配置类中完成,具体要怎么做,我们来看看,stock模块之前写过quartz,我们就把config类建在quartz包下:

package com.codingfire.cloud.stock.quartz;

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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

// SpringBoot整合RabbitMQ之后
// 这些配置信息要保存在Spring容器中,所以这些配置也要交给SpringBoot管理
@Configuration
public class RabbitMQConfig 
    // 声明需要使用的交换机\\路由Key\\队列的名称
    public static final String STOCK_EX="stock_ex";
    public static final String STOCK_ROUT="stock_rout";
    public static final String STOCK_QUEUE="stock_queue";

    // 声明交换机,需要几个声明几个,这里就一个
    // 方法中实例化交换机对象,确定名称,保存到Spring容器
    @Bean
    public DirectExchange stockDirectExchange()
        return new DirectExchange(STOCK_EX);
    

    // 声明队列,需要几个声明几个,这里就一个
    // 方法中实例化队列对象,确定名称,保存到Spring容器
    @Bean
    public Queue stockQueue()
        return new Queue(STOCK_QUEUE);
    

    // 声明路由Key(交换机和队列的关系),需要几个声明几个,这里就一个
    // 方法中实例化路由Key对象,确定名称,保存到Spring容器
    @Bean
    public Binding stockBinding()
        return BindingBuilder.bind(stockQueue()).to(stockDirectExchange())
                .with(STOCK_ROUT);
    

都是比较固定的模版,直接用就行。

RabbitMQ消息的发送

发送消息,我们可以在QuartzJob类中进行:

package com.codingfire.cloud.stock.quartz;

import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.LocalDateTime;

public class QuartzJob implements Job 

    // RabbitTemplate就是amqp框架提供的发送消息的对象
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException 
        //输出当前时间
        System.out.println("--------------"+ LocalDateTime.now() +"---------------");
        // 先简单的发送一串文字
        rabbitTemplate.convertAndSend(RabbitMQConfig.STOCK_EX,
                RabbitMQConfig.STOCK_ROUT,"黄河之水天上来,奔流到海不复还。");

    

我们在讲解Quartz时只用这个类每隔10s打印出当前时间,现在我们在打印时间的同时,再用RabbitMQ发送一串文字,rabbitTemplate和使用redi很像,不同的是,这里是通过template指定发送的交换机和路由。 

RabbitMQ消息的接收

我们在quartz包下再建一个用于接收消息的类:

package com.codingfire.cloud.stock.quartz;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

// 这个对象也是需要交由Spring容器管理的,才能实现监听Spring容器中保存的队列的效果
@Component
// 和Kafka不同的是Kafka在一个方法上声明监听器
// 而RabbitMQ是在类上声明,监听具体的队列名称
@RabbitListener(queues = RabbitMQConfig.STOCK_QUEUE)
public class RabbitMQConsumer 

    // 监听了类,但是运行代码的一定是个方法
    // 框架要求这个类中只允许一个方法包含下面这个注解
    // 表示这个方法是处理消息的方法
    // 方法的参数就是消息的值
    @RabbitHandler
    public void process(String str)
        System.out.println("接收到的消息为:"+str);
    



里面的方法名可以是自定义的,只要类上指定队列名称就可以接收,类中注释认真看,要记住。 

测试

做完这些,下面就是启动并测试的环节,由于此微服务模块配置了nacos和seata,所以这两个服务是要起的,接着就是RabbitMQ,上面已经教大家启动过了,保证是运行状态:然后运行stock的启动文件,在控制台查看消息发送接收的情况,由于博主还依赖了数据库,此处还需要启动mysql服务,大家根据自身情况启动服务,然后再次运行:

 

可以看到我们的消息收发是正常的,测试成功,同时能看到我们做Quartz时,减少库存的操作也在运行。这就是 RabbitMQ的基本使用。

在RabbitMQ后台,也能看到我们注册的交换机和消息队列等信息:

感兴趣的可以去看看。 

结语

又到了说再见的时候,RabbitMQ和Kafka一样博大精深,这里只是基本的使用,你只要知道,他们是用来收发消息的,且不需要关心什么时候完成,你可以认为他们是在系统不忙的时候才去做,这样可以降低服务器压力,另外还能实时监控:

是不是很方便呢?赶快到自己的项目中使用吧。 

以上是关于Java开发 - 消息队列之RabbitMQ初体验的主要内容,如果未能解决你的问题,请参考以下文章

Java开发 - ELK初体验

Java消息队列--ActiveMq 初体验

Java消息队列--ActiveMq 初体验

php消息队列之 think queue消息队列初体验

php rabbitmq的开发体验

Java教程之RabbitMQ介绍