Spring 集成 RabbitMQ 与其概念，消息持久化，ACK机制

2024-06-24 07:58
文章标签 spring rabbitmq ack 集成 持久 机制 概念 消息 java

本文主要是介绍Spring 集成 RabbitMQ 与其概念，消息持久化，ACK机制，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

目录

  • RabbitMQ 概念
  • exchange交换机机制
    • 什么是交换机
    • binding？
    • Direct Exchange交换机
    • Topic Exchange交换机
    • Fanout Exchange交换机
    • Header Exchange交换机
  • RabbitMQ 的 Hello - Demo（springboot实现）
  • RabbitMQ 的 Hello Demo（spring xml实现）
  • RabbitMQ 在生产环境下运用和出现的问题
    • Spring RabbitMQ 注解
    • 消息的 JSON 传输
    • 消息持久化，断线重连，ACK。

RabbitMQ 概念

RabbitMQ 即一个消息队列，_主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。_RabbitMQ使用的是AMQP协议，它是一种二进制协议。默认启动端口 5672。

在 RabbitMQ 中，如下图结构：

rabbitmq

  • 左侧 P 代表 生产者，也就是往 RabbitMQ 发消息的程序。
  • 中间即是 RabbitMQ，_其中包括了 交换机 和 队列。_
  • 右侧 C 代表 消费者，也就是往 RabbitMQ 拿消息的程序。

那么，_其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。_

  • 虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，_用户只能在虚拟主机的粒度进行权限控制。_ 因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。
  • 交换机：_Exchange 用于转发消息，但是它不会做存储_ ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。
    • 这里有一个比较重要的概念：**路由键 ** 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
  • 绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

exchange交换机机制

什么是交换机

rabbitmq的message model实际上消息不直接发送到queue中，中间有一个exchange是做消息分发，producer甚至不知道消息发送到那个队列中去。因此，当exchange收到message时，必须准确知道该如何分发。是append到一定规则的queue，还是append到多个queue中，还是被丢弃？这些规则都是通过exchagne的4种type去定义的。

The core idea in the messaging model in RabbitMQ is that the producer never sends any messages directly to a queue. Actually, quite often the producer doesn't even know if a message will be delivered to any queue at all.

Instead, the producer can only send messages to an exchange. An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.

exchange是一个消息的agent，每一个虚拟的host中都有定义。它的职责是把message路由到不同的queue中。

binding？

exchange和queue通过routing-key关联，这两者之间的关系是就是binding。如下图所示,X表示交换机，红色表示队列，交换机通过一个routing-key去binding一个queue，routing-key有什么作用呢？看Direct exchange类型交换机。

Directed Exchange

路由键exchange，该交换机收到消息后会把消息发送到指定routing-key的queue中。那消息交换机是怎么知道的呢？其实，producer deliver消息的时候会把routing-key add到 message header中。routing-key只是一个messgae的attribute。

A direct exchange delivers messages to queues based on a message routing key. The routing key is a message attribute added into the message header by the producer. The routing key can be seen as an "address" that the exchange use to decide how to route the message. A message goes to the queue(s) whose binding key exactly matches the routing key of the message.

Default Exchange
这种是特殊的Direct Exchange，是rabbitmq内部默认的一个交换机。该交换机的name是空字符串，所有queue都默认binding 到该交换机上。所有binding到该交换机上的queue，routing-key都和queue的name一样。

Topic Exchange

通配符交换机，exchange会把消息发送到一个或者多个满足通配符规则的routing-key的queue。其中_表号匹配一个word，#匹配多个word和路径，路径之间通过.隔开。如满足a._.c的routing-key有a.hello.c；满足#.hello的routing-key有a.b.c.helo。

Fanout Exchange

扇形交换机，该交换机会把消息发送到所有binding到该交换机上的queue。这种是publisher/subcribe模式。用来做广播最好。
所有该exchagne上指定的routing-key都会被ignore掉。

The fanout copies and routes a received message to all queues that are bound to it regardless of routing keys or pattern matching as with direct and topic exchanges. Keys provided will simply be ignored.

Header Exchange

设置header attribute参数类型的交换机。

RabbitMQ 的 Hello Demo

安装就不说了，建议按照官方文档上做。先贴代码，稍后解释，代码如下：

配置 交换机，队列，交换机与队列的绑定，消息监视容器：

@Configuration
@Data
public class RabbitMQConfig {final static String queueName = "spring-boot";@BeanQueue queue() {return new Queue(queueName, false);}@BeanTopicExchange exchange() {return new TopicExchange("spring-boot-exchange");}@BeanBinding binding(Queue queue, TopicExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with(queueName);}@BeanSimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(queueName);container.setMessageListener(listenerAdapter);return container;}@BeanReceiver receiver() {return new Receiver();}@BeanMessageListenerAdapter listenerAdapter(Receiver receiver) {return new MessageListenerAdapter(receiver, "receiveMessage");}
}

配置接收信息者（即消费者）：

public class Receiver {private CountDownLatch latch = new CountDownLatch(1);public void receiveMessage(String message) {System.out.println("Received <" + message + ">");latch.countDown();}public CountDownLatch getLatch() {return latch;}
}

配置发送信息者（即生产者）：

@RestController
public class Test {@AutowiredRabbitTemplate rabbitTemplate;@RequestMapping(value = "/test/{abc}",method = RequestMethod.GET)public String test(@PathVariable(value = "abc") String abc){rabbitTemplate.convertAndSend("spring-boot", abc + " from RabbitMQ!");return  "abc";}
}

以上便可实现一个简单的 RabbitMQ Demo，具体代码在：点这里

那么，这里，分为三个部分分析：发消息，交换机队列，收消息。

  • 对于发送消息：我们一般可以使用 RabbitTemplate，这个是 Spring 封装给了我们，便于我们发送信息，我们调用 rabbitTemplate.convertAndSend("spring-boot", xxx); 即可发送信息。
  • 对于交换机队列：如上代码，我们需要配置交换机 TopicExchange，配置队列 Queue，并且配置他们之间的绑定 Binding
  • 对于接受消息：首先需要创建一个消息监听容器，然后把我们的接受者注册到该容器中，这样，队列中有信息，那么就会调用接收者的对应的方法。如上代码 container.setMessageListener(listenerAdapter); 其中，MessageListenerAdapter 可以看做是 我们接收者的一个包装类，new MessageListenerAdapter(receiver, "receiveMessage"); 指明了如果有消息来，那么调用接收者哪个方法进行处理。

RabbitMQ 的 Hello Demo（spring xml实现）

spring xml方式实现RabbitMQ简单，可读性较好，配置简单，配置和实现如下所示。

上文已经讲述了rabbitmq的配置，xml方式通过properites文件存放用户配置信息：

mq.host=127.0.0.1
mq.username=guest
mq.password=guest
mq.port=5672

配置application-mq.xml配置文件，声明连接、交换机、queue以及consumer监听。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" ><description>rabbitmq 连接服务配置</description><!-- 连接配置 --><context:property-placeholder location="classpath:mq.properties" /><rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}"/><rabbit:admin connection-factory="connectionFactory"/><!-- spring template声明--><rabbit:template exchange="amqpExchange" id="amqpTemplate"  connection-factory="connectionFactory" /><!--申明queue--><rabbit:queue id="test_queue_key" name="test_queue_key" durable="true" auto-delete="false" exclusive="false" /><!--申明exchange交换机并绑定queue--><rabbit:direct-exchange name="amqpExchange" durable="true" auto-delete="false" id="amqpExchange"><rabbit:bindings><rabbit:binding queue="test_queue_key" key="test_queue_key"/></rabbit:bindings></rabbit:direct-exchange><!--consumer配置监听--><bean id="reveiver" class="com.demo.mq.receive.Reveiver" /><rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"><rabbit:listener queues="test_queue_key" ref="reveiver" method="receiveMessage"/></rabbit:listener-container>
</beans>

上述代码中，引入properties文件就不多说了。

<rabbit:connection-factory>标签声明创建connection的factory工厂。

<rabbit-template>声明spring template，和上文spring中使用template一样。template可声明exchange。

<rabbit:queue>声明一个queue并设置queue的配置项，直接看标签属性就可以明白queue的配置项。

<rabbit:direct-exchange>声明交换机并绑定queue。

<rabbit:listener-container>申明监听container并配置consumer和监听routing-key。

剩下就简单了，application-context.xml中把rabbitmq配置import进去。

<?xml version="1.0" encoding="UTF-8"?>
<beansxmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:task="http://www.springframework.org/schema/task"xmlns:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsdhttp://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsdhttp://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd"><context:component-scan base-package="com.demo.**" /><import resource="application-mq.xml" />
</beans>

Producer实现，发送消息还是使用template的convertAndSend() deliver消息。

@Service
public class Producer {@Autowiredprivate AmqpTemplate amqpTemplate;private final static Logger logger = LoggerFactory.getLogger(Producer.class);public void sendDataToQueue(String queueKey, Object object) {try {amqpTemplate.convertAndSend(queueKey, object);} catch (Exception e) {e.printStackTrace();logger.error("exeception={}",e);}}
}

配置consumer

package com.demo.mq.receive;import org.springframework.stereotype.Service;
import java.util.concurrent.CountDownLatch;@Service
public class Reveiver {private CountDownLatch latch = new CountDownLatch(1);public void receiveMessage(String message) {System.out.println("reveice msg=" + message.toString());latch.countDown();}
}

测试deliver消息

Controller
@RequestMapping("/demo/")
public class TestController {private final static Logger logger = LoggerFactory.getLogger(TestController.class);@Resourceprivate Producer producer;@RequestMapping("/test/{msg}")public String send(@PathVariable("msg") String msg){logger.info("#TestController.send#abc={msg}", msg);System.out.println("msg="+msg);producer.sendDataToQueue("test_queue_key",msg);return "index";}
}

RabbitMQ 在生产环境下运用和出现的问题

在生产环境中，由于 Spring 对 RabbitMQ 提供了一些方便的注解，所以首先可以使用这些注解。例如：

  • @enablerabbit：@enablerabbit 和 @configuration 注解在一个类中结合使用，如果该类能够返回一个 RabbitListenerContainerFactory 类型的 bean，那么就相当于能够把该终端（消费端）和 RabbitMQ 进行连接。Ps：（生成端不是通过 RabbitListenerContainerFactory 来和 RabbitMQ 连接，而是通过 RabbitTemplate ）
  • @rabbitlistener：当对应的队列中有消息的时候，该注解修饰下的方法会被执行。
  • @rabbithandler：接收者可以监听多个队列，不同的队列消息的类型可能不同，该注解可以使得不同的消息让不同方法来响应。

具体这些注解的使用，可以参考这里的代码：点这里

首先，生产环境下的 RabbitMQ 可能不会在生产者或者消费者本机上，所以需要重新定义 ConnectionFactory，即：

@Bean
ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);connectionFactory.setUsername(userName);connectionFactory.setPassword(password);connectionFactory.setVirtualHost(vhost);return connectionFactory;
}

这里，可以重新设置需要连接的 RabbitMQ 的 ip，端口，虚拟主机，用户名，密码。

然后，可以先从生产端考虑，生产端需要连接 RabbitMQ，那么可以通过 RabbitTemplate 进行连接。 Ps：（RabbitTemplate 用于生产端发送消息到交换机中），如下代码：

@Bean(name="myTemplate")
RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate template = new RabbitTemplate(connectionFactory);template.setMessageConverter(integrationEventMessageConverter());template.setExchange(exchangeName);return template;
}

在该代码中，new RabbitTemplate(connectionFactory); 设置了生产端连接到RabbitMQ，template.setMessageConverter(integrationEventMessageConverter()); 设置了 生产端发送给交换机的消息是以什么格式的，在 integrationEventMessageConverter() 代码中：

public MessageConverter integrationEventMessageConverter() {Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();return messageConverter;
}

如上 Jackson2JsonMessageConverter 指明了 JSON。上述代码的最后 template.setExchange(exchangeName); 指明了 要把生产者要把消息发送到哪个交换机上。

有了上述，那么，我们即可使用 rabbitTemplate.convertAndSend("spring-boot", xxx); 发送消息，xxx 表示任意类型，因为上述的设置会帮我们把这些类型转化成 JSON 传输。

接着，生产端发送我们说过了，那么现在可以看看消费端：

对于消费端，我们可以只创建 SimpleRabbitListenerContainerFactory，它能够帮我们生成 RabbitListenerContainer，然后我们再使用 @rabbitlistener 指定接收者收到信息时处理的方法。

@Bean(name="myListenContainer")
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setMessageConverter(integrationEventMessageConverter());factory.setConnectionFactory(connectionFactory());return factory;
}

这其中 factory.setMessageConverter(integrationEventMessageConverter()); 指定了我们接受消息的时候，以 JSON 传输的消息可以转换成对应的类型传入到方法中。例如：

@Slf4j
@Component
@RabbitListener(containerFactory = "helloRabbitListenerContainer",queues = "spring-boot")
public class Receiver {@RabbitHandlerpublic void receiveTeacher(Teacher teacher) {log.info("##### = {}",teacher);}
}

可能出现的问题：

消息持久化

在生产环境中，我们需要考虑万一生产者挂了，消费者挂了，或者 rabbitmq 挂了怎么样。一般来说，如果生产者挂了或者消费者挂了，其实是没有影响，因为消息就在队列里面。那么万一 rabbitmq 挂了，之前在队列里面的消息怎么办，其实可以做消息持久化，RabbitMQ 会把信息保存在磁盘上。

做法是可以先从 Connection 对象中拿到一个 Channel 信道对象，然后再可以通过该对象设置 消息持久化。

生产者或者消费者断线重连

这里 Spring 有自动重连机制。

ACK 确认机制

每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中，Consumer出错了，异常退出了，而数据还没有处理完成，那么 非常不幸，这段数据就丢失了。因为我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完 成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。

如果一个Consumer异常退出了，它处理的数据能够被另外的Consumer处理，这样数据在这种情况下就不会丢失了（注意是这种情况下）。
为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。
如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。

个人对 RabbitMQ ACK 的一些疑问，求助：点这里

总结

  1. RabbitMQ 作用：异步，解耦，缓冲，消息分发。
  2. RabbitMQ 主要分为3个部分，生产者，交换机和队列，消费者。
  3. 需要注意消息持久化，目的为了防止 RabbitMQ 宕机；考虑 ACK 机制，目的为了如果消费者对消息的处理失败了，那么后续要如何处理。

写在最后

  1. 写出来，说出来才知道对不对，知道不对才能改正，改正了才能成长。
  2. 在技术方面，希望大家眼里都容不得沙子。如果有不对的地方或者需要改进的地方希望可以指出，万分感谢。

这篇关于Spring 集成 RabbitMQ 与其概念，消息持久化，ACK机制的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1089567。

相关文章

Java五子棋之坐标校正

Java五子棋之坐标校正

上篇针对了Java项目中的解构思维，在这篇内容中我们不妨从整体项目中拆解拿出一个非常重要的五子棋逻辑实现：坐标校正，我们如何使漫无目的鼠标点击变得有序化和可控化呢？ 目录 一、从鼠标监听到获取坐标 1.MouseListener和MouseAdapter 2.mousePressed方法 二、坐标校正的具体实现方法 1.关于fillOval方法 2.坐标获取 3.坐标转换 4.坐
阅读更多...
Spring Cloud：构建分布式系统的利器

Spring Cloud：构建分布式系统的利器

引言 在当今的云计算和微服务架构时代，构建高效、可靠的分布式系统成为软件开发的重要任务。Spring Cloud 提供了一套完整的解决方案，帮助开发者快速构建分布式系统中的一些常见模式（例如配置管理、服务发现、断路器等）。本文将探讨 Spring Cloud 的定义、核心组件、应用场景以及未来的发展趋势。 什么是 Spring Cloud Spring Cloud 是一个基于 Spring
阅读更多...
Javascript高级程序设计(第四版)--学习记录之变量、内存

Javascript高级程序设计(第四版)--学习记录之变量、内存

原始值与引用值 原始值：简单的数据即基础数据类型，按值访问。 引用值：由多个值构成的对象即复杂数据类型，按引用访问。 动态属性 对于引用值而言，可以随时添加、修改和删除其属性和方法。 let person = new Object();person.name = 'Jason';person.age = 42;console.log(person.name,person.age);//'J
阅读更多...
java8的新特性之一（Java Lambda表达式）

java8的新特性之一（Java Lambda表达式）

1：Java8的新特性 Lambda 表达式： 允许以更简洁的方式表示匿名函数（或称为闭包）。可以将Lambda表达式作为参数传递给方法或赋值给函数式接口类型的变量。 Stream API： 提供了一种处理集合数据的流式处理方式，支持函数式编程风格。 允许以声明性方式处理数据集合（如List、Set等）。提供了一系列操作，如map、filter、reduce等，以支持复杂的查询和转
阅读更多...
Java面试八股之怎么通过Java程序判断JVM是32位还是64位

Java面试八股之怎么通过Java程序判断JVM是32位还是64位

怎么通过Java程序判断JVM是32位还是64位 可以通过Java程序内部检查系统属性来判断当前运行的JVM是32位还是64位。以下是一个简单的方法： public class JvmBitCheck {public static void main(String[] args) {String arch = System.getProperty("os.arch");String dataM
阅读更多...
详细分析Springmvc中的@ModelAttribute基本知识（附Demo）

详细分析Springmvc中的@ModelAttribute基本知识（附Demo）

目录 前言1. 注解用法1.1 方法参数1.2 方法1.3 类 2. 注解场景2.1 表单参数2.2 AJAX请求2.3 文件上传 3. 实战4. 总结 前言 将请求参数绑定到模型对象上，或者在请求处理之前添加模型属性 可以在方法参数、方法或者类上使用 一般适用这几种场景： 表单处理：通过 @ModelAttribute 将表单数据绑定到模型对象上预处理逻辑：在请求处理之前
阅读更多...
eclipse运行springboot项目，找不到主类

eclipse运行springboot项目，找不到主类

解决办法尝试了很多种，下载sts压缩包行不通。最后解决办法如图： help--->Eclipse Marketplace--->Popular--->找到Spring Tools 3---->Installed。
阅读更多...
JAVA读取MongoDB中的二进制图片并显示在页面上

JAVA读取MongoDB中的二进制图片并显示在页面上

1：Jsp页面： <td><img src="${ctx}/mongoImg/show"></td> 2：xml配置： <?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001
阅读更多...
持久层 技术选型如何决策？JPA，Hibernate，ibatis(mybatis)

持久层 技术选型如何决策？JPA，Hibernate，ibatis(mybatis)

转自：http://t.51jdy.cn/thread-259-1-1.html 持久层 是一个项目 后台 最重要的部分。他直接 决定了 数据读写的性能，业务编写的复杂度，数据结构(对象结构)等问题。 因此 架构师在考虑 使用那个持久层框架的时候 要考虑清楚。 选择的 标准： 1，项目的场景。 2，团队的技能掌握情况。 3，开发周期(开发效率)。 传统的 业务系统，通常业
阅读更多...
Java面试题：通过实例说明内连接、左外连接和右外连接的区别

Java面试题：通过实例说明内连接、左外连接和右外连接的区别

在 SQL 中，连接（JOIN）用于在多个表之间组合行。最常用的连接类型是内连接（INNER JOIN）、左外连接（LEFT OUTER JOIN）和右外连接（RIGHT OUTER JOIN）。它们的主要区别在于它们如何处理表之间的匹配和不匹配行。下面是每种连接的详细说明和示例。 表示例 假设有两个表：Customers 和 Orders。 Customers CustomerIDCus
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2