Spring RabbitMQ那些事(3-消息可靠传输和订阅)

发布时间:2024年01月20日

一、序言

在有些业务场景中,消息是不能丢的,比如分布式事务资金动账,出账方扣款,那么入账方就一定要收款。以前写了一篇分布式事务的文章,里面的跨地区转账就是一个实际案例。

消息是有可能丢的,比如生产者在发送消息时broker服务挂了,消息没有来得及落盘,这时消息就彻底丢了。

保证MQ消息可靠传输主要有两个方面,一方面是消息生产者确保消息一定发送成功,另一方面是消费者确保消息一定被处理。


二、生产者确保消息发送成功

1、为什么需要Publisher Confirms

在Spring AMQP中AmqpTemplate的实现RabbitTemplate已经支持 Publisher Confirms and Returns,所谓的publisher confirms意思就是消息发布者确认消息是否已经被发送。

在RabbitMQ官方文档描述中,持久化的消息在Broker重启时也是应该存活的,这里的词用的是应该,因为消息有可能在落地磁盘前Broker就挂了,导致消息丢失。

最直接的解决方案是通过事务,但是通过事务有两个问题:

  • 事务阻塞:发布者必须等待Broker处理完每条消息。
  • 事务很重:每次提交都会要求触发fsync(),强制磁盘,这个过程需要花很长的时间。

备注:在RabbitMQ官方测试中,通过事务去保证,发布10000条消息需要花至少4分钟的时间。

在这里插入图片描述

而通过Publisher Confirm机制,一旦Broker处理完就会确认消息,而且这个过程是异步的,生产者可以流式发布消息,不需要等待Broker,并且Broker会批量高效将消息落盘。

2、哪些消息会被确认处理成功

当Broker确认消息时,会通知消息发布者消息是否被成功处理,成功处理的基本规则如下:

  • 无法路由的mandatory(必须有符合条件的队列)和immediate(必须有消费者在线)类型在被basic.return后会被确认。
  • 非持久化消息在入队时会被确认。
  • 持久化消息当持久化到磁盘或者被消费者消费时会被确认。

三、消费者保证消息被处理

消费者端确保消息消费很简单,关闭消息自动确认就好,开启消息手动确认。当然有些场景消息只能被处理一次,可以通过分布式锁来实现。


四、Spring RabbitMQ支持代码示例

1、 application.yml

server:
  port: 8080
spring:
  rabbitmq:
    addresses: localhost:5672
    username: admin
    password: admin
    virtual-host: /
    publisher-returns: true
    publisher-confirm-type: correlated
    listener:
      type: simple
      simple:
        acknowledge-mode: manual
        concurrency: 5
        max-concurrency: 20
        prefetch: 5
    template:
      mandatory: true

备注:

  • 这里一定要设置spring.rabbitmq.publisher-returnstrue,并且设置spring.rabbitmq.publisher-confirm-typecorrelated,同时设置spring.rabbitmq.template.mandatorytrue
  • 上面我们将消费者的确认模式改为了手动确认

2、RabbigtMQ配置

@Configuration
public class RabbitReliableTransportConfig {

	/**
	 * RabbitTemplate消息转换器配置,自动将对象转换为json字符串
	 *
	 * @return
	 */
	@Bean
	public MessageConverter jackson2JsonMessageConverter() {
		Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();
		messageConverter.setClassMapper(new DefaultJackson2JavaTypeMapper());
		return messageConverter;
	}

	@Bean
	public Queue reliableQueue() {
		return QueueBuilder.durable("reliable-queue").build();
	}
}

3、可靠生产者配置

@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMqReliableProducer {

	private final RabbitTemplate rabbitTemplate;

	public void sendReliableMsg(String body) {
		// 发送可靠消息
		ReliableMsgDTO reliableMsgDTO = ReliableMsgDTO.builder().body(body).build();
		CorrelationData correlationData = new CorrelationData();
		rabbitTemplate.convertAndSend("reliable-queue", reliableMsgDTO, correlationData);

		// 发送确认逻辑
		CompletableFuture<Confirm> future = correlationData.getFuture().completable();
		future.whenComplete((confirm, throwable) -> {
			if (confirm.isAck()) {
				log.info("消息已经被成功发送, 消息内容:{}", JSON.toJSONString(reliableMsgDTO));
				return;
			}

			log.warn("消息发送未成功发送, 原因:{}, 消息内容:{}", confirm.getReason(), JSON.toJSONString(reliableMsgDTO), throwable);
			// 5秒后再发送
			LockSupport.parkNanos(5 * 1000 * 1000 * 1000L);
			rabbitTemplate.convertSendAndReceive(reliableMsgDTO, correlationData);
		});
	}
}

4、可靠消费者配置

@Slf4j
@Component
public class RabbitMQReliableConsumer {

	@RabbitListener(queues = "reliable-queue")
	public void handleMsgFromQueue(ReliableMsgDTO reliableMsgDTO, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
		channel.basicAck(tag, false);
		// channel.basicNack(tag, false, false);
		log.info("Message received from queue, message body: {}", JSON.toJSONString(reliableMsgDTO));
	}
}

备注:这里我们开启了消息的手动确认,如果消息处理失败没有确认,那么消息将会在下次消费者参加连接时再次被投递。

5、测试用例

测试结果如下,每当消息发送至Broker成功后会触发回调,如果消息发送失败将会触发重新发送。

2024-01-20 18:13:11.399  INFO 12316 --- [78.107.127:5672] c.u.r.i.p.RabbitMqReliableProducer       : 消息已经被成功发送, 消息内容:{"body":"hello"}
2024-01-20 18:13:11.399  INFO 12316 --- [ntContainer#0-5] c.u.r.i.c.RabbitMQReliableConsumer       : Message received from queue, message body: {"body":"hello"}

在这里插入图片描述

文章来源:https://blog.csdn.net/lingbomanbu_lyl/article/details/134674676
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。