RabbitMQ入门指南(三):Java入门示例

简介: RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了AMQP、Spring AMQP和使用SpringAMQP实现对RabbitMQ的消息收发等内容。

一、AMQP协议

1.AMQP

全称为Advanced Message Queuing Protocol,是一种用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关,更符合微服务中独立性的要求。通过AMQP,不同的应用程序可以在不改变各自实现方式的情况下进行跨平台、跨语言的消息通信。

AMQP协议定义了消息的传输方式和消息的元数据,例如消息的发送者、接收者、消息体、消息类型等。这些元数据可以帮助应用程序对消息进行正确的处理。

2.Spring AMQP

在Spring框架中,有一个Spring AMQP的项目,它基于AMQP协议定义了一套API规范,提供了模板来发送和接收消息。这个项目包含两部分,其中spring-amqp是基础抽象,而spring-rabbit是底层的默认实现。

Spring AMQP通过提供模板和抽象层,简化了应用程序与RabbitMQ的交互。它提供了一组易于使用的API,用于发送和接收消息。这些API可以帮助开发人员更专注于业务逻辑,而不是消息的发送和接收细节。

spring-rabbit是Spring AMQP的一部分,它基于RabbitMQ实现了AMQP协议。spring-rabbit提供了对RabbitMQ的封装,使开发人员可以通过简单的配置和API调用与RabbitMQ进行交互。

Spring AMQP 主要功能:

  • 自动声明和配置队列、交换机及其绑定关系:通过简化队列和交换器的创建和管理过程,Spring AMQP 帮助开发人员专注于实现业务逻辑,而不是手动配置消息中间件。
  • 基于注解的监听器模式,实现异步消息接收:通过注解,Spring AMQP 可以自动将方法与特定的队列或交换机绑定,从而实现异步接收和处理消息。这种模式提高了应用程序的响应性能和吞吐量。
  • 封装了 RabbitTemplate 工具,用于发送消息:RabbitTemplate 是 RabbitMQ 的核心类之一,用于发送和接收消息。Spring AMQP 提供了对这个工具的封装,使得开发人员可以方便地使用它来发送消息。

官方文档:

https://spring.io/projects/spring-amqp

二、使用Spring AMQP实现对RabbitMQ的消息收发

1.案例准备阶段

项目结构如下:

image.gif 项目结构介绍:

  • mq-demo:父工程,管理项目依赖
  • publisher:消息的发送者
  • consumer:消息的消费者


在父工程引入spring-amqp依赖:

<!--AMQP依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

image.gif

项目完整依赖如下:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.rye.demo</groupId>
    <artifactId>mq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>publisher</module>
        <module>consumer</module>
    </modules>
    <packaging>pom</packaging>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.15</version>
        <relativePath/>
    </parent>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--AMQP依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--单元测试-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <!--Jackson-->
        <dependency>
            <groupId>com.fasterxml.jackson.dataformat</groupId>
            <artifactId>jackson-dataformat-xml</artifactId>
        </dependency>
    </dependencies>
</project>

image.gif

在application.yml中配置RabbitMQ服务端信息(每个微服务都需要配置):

spring:
  rabbitmq:
    host: 10.0.0.100
    port: 5672
    virtual-host: /demo
    username: user
    password: 123456

image.gif

2.入门案例(无交换机)

案例模型:

image.gif 在RabbitMQ管理控制台新建队列:

image.gif 查看新建结果:

image.gif 在publisher服务中编写测试类,并利用RabbitTemplate实现消息发送

@Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void testSendMessage2Queue() {
        // 队列名称
        String queueName = "demo.queue";
        // 消息
        String msg = "First demo";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, msg);
    }

image.gif

运行测试用例,查看结果:

image.gif


在consumer服务中新建一个类实现消息接收

@Component
public class MqListener {
    @RabbitListener(queues = "demo.queue")
    public void listenSimpleQueue(String msg) {
        System.out.println("消息:" + msg);
    }
}

image.gif

启动consumer服务,查看消息(一旦监听的队列中有了消息,就会推送给当前服务):


3.任务模型案例(Work Queues)

让多个消费者绑定一个队列,共同消费队列中的消息。

案例模型:


在RabbitMQ管理控制台新建队列:


查看新建结果:


在publisher服务中的测试类添加一个测试方法(通过循环发送,模拟大量消息堆积现象 ):

@Test
    void testWorkQueue() throws InterruptedException {
        String queueName = "work.queue";
        for (int i = 1; i <= 50; i++) {
            String msg = "Work Queues " + i;
            rabbitTemplate.convertAndSend(queueName, msg);
            Thread.sleep(20);
        }
    }

image.gif

在consumer服务的类中添加2个新的方法,模拟多个消费者绑定同一个队列 :

@RabbitListener(queues = "work.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收work.queue消息:" + msg);
    }
    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2接收work.queue消息:" + msg);
    }

image.gif

运行结果:


修改consumer服务类中的方法:

  • 消费者1 sleep了20毫秒,相当于每秒钟处理50个消息
  • 消费者2 sleep了200毫秒,相当于每秒处理5个消息
@RabbitListener(queues = "work.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收work.queue消息:" + msg);
        Thread.sleep(20);
    }
    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2接收work.queue消息:" + msg);
        Thread.sleep(200);
    }

image.gif

重启后查看运行结果:


以上结果表明:默认情况下,RabbitMQ会将消息依次轮询投递给绑定在队列上的每一个消费者,并没有考虑到消费者是否已经处理完消息,可能出现消息堆积。

修改consumer服务的application.yml,设置preFetch值为1,确保同一时刻最多投递给消费者1条消息(每次只能获取一条消息,处理完成才能获取下一个消息):

spring:
  rabbitmq:
    host: 10.0.0.100
    port: 5672
    virtual-host: /demo
    username: user
    password: 123456
    listener:
      simple:
        prefetch: 1

image.gif

重启后查看运行结果:



总结

RabbitMQ是一个开源的消息队列软件,旨在提供可靠的消息传递和消息队列功能。本文主要介绍了AMQP、Spring AMQP和使用Spring AMQP实现对RabbitMQ的消息收发等内容,希望对大家有所帮助。

相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
4天前
|
网络协议 安全 Java
Java网络编程入门涉及TCP/IP协议理解与Socket通信。
【6月更文挑战第21天】Java网络编程入门涉及TCP/IP协议理解与Socket通信。TCP/IP协议包括应用层、传输层、网络层和数据链路层。使用Java的`ServerSocket`和`Socket`类,服务器监听端口,接受客户端连接,而客户端连接指定服务器并交换数据。基础示例展示如何创建服务器和发送消息。进阶可涉及多线程、NIO和安全传输。学习这些基础知识能助你构建网络应用。
14 1
|
2天前
|
Java Android开发
Java Socket编程示例:服务器开启在8080端口监听,接收客户端连接并打印消息。
【6月更文挑战第23天】 Java Socket编程示例:服务器开启在8080端口监听,接收客户端连接并打印消息。客户端连接服务器,发送&quot;Hello, Server!&quot;后关闭。注意Android中需避免主线程进行网络操作。
12 4
|
15小时前
|
Java
java代码示例和常用代码解读
java代码示例和常用代码解读
|
1天前
|
监控 安全 Java
构建Java版监控系统:从入门到精通
构建Java版监控系统:从入门到精通
5 1
|
1天前
|
Java
轻松入门Java中的Lambda函数
轻松入门Java中的Lambda函数
|
5天前
|
Java
死锁是线程间争夺资源造成的无限等待现象,Java示例展示了两个线程各自持有资源并等待对方释放,导致死锁。`
【6月更文挑战第20天】死锁是线程间争夺资源造成的无限等待现象,Java示例展示了两个线程各自持有资源并等待对方释放,导致死锁。`volatile`保证变量的可见性和部分原子性,确保多线程环境中值的即时更新。与`synchronized`相比,`volatile`作用于单个变量,不保证原子操作,同步范围有限,但开销较小。`synchronized`提供更全面的内存语义,保证原子性和可见性,适用于复杂并发控制。
13 3
|
4天前
|
存储 算法 Java
Java查找算法概览:二分查找适用于有序数组,通过比较中间元素缩小搜索范围;哈希查找利用哈希函数快速定位,示例中使用HashMap存储键值对,支持多值关联。
【6月更文挑战第21天】Java查找算法概览:二分查找适用于有序数组,通过比较中间元素缩小搜索范围;哈希查找利用哈希函数快速定位,示例中使用HashMap存储键值对,支持多值关联。简单哈希表实现未涵盖冲突解决和删除操作。
12 1
|
5天前
|
Java 开发者
告别单线程时代!Java 多线程入门:选继承 Thread 还是 Runnable?
【6月更文挑战第19天】在Java中,面对多任务需求时,开发者可以选择继承`Thread`或实现`Runnable`接口来创建线程。`Thread`继承直接但限制了单继承,而`Runnable`接口提供多实现的灵活性和资源共享。多线程能提升CPU利用率,适用于并发处理和提高响应速度,如在网络服务器中并发处理请求,增强程序性能。不论是选择哪种方式,都是迈向高效编程的重要一步。
|
7天前
|
存储 安全 Java
Java Queue:从入门到精通,一篇文章就够了!
【6月更文挑战第18天】Java集合框架中的队列Queue遵循FIFO原则,用于存储和管理元素。从创建队列(如LinkedList示例)到移除元素(remove和poll方法),再到不同实现类(如ArrayDeque和ConcurrentLinkedQueue),队列在多线程、任务调度等场景中广泛应用。自定义队列如LimitedQueue展示如何限制容量。了解并熟练使用队列能提升程序性能和可读性。队列,是高效编程的关键工具。
|
7天前
|
安全 Java 索引
Java List:从入门到精通,一篇文章就够了!
【6月更文挑战第17天】Java List是有序元素集合,支持索引访问、添加、删除和修改。从ArrayList、LinkedList到Vector,各种实现满足不同场景需求。使用add()添加元素,get()获取,set()修改,remove()删除。遍历可用for-each或Iterator,subList()创建子集。注意线程安全,可选synchronizedList()、Vector或CopyOnWriteArrayList。理解List的基本操作和特性,能提升编程效率。