SpringBoot가 RabbitMQ를 통합하여 지연 대기열을 구현하는 방법
- WBOY앞으로
- 2023-05-16 20:31:10878검색
메시지가 손실되지 않도록 하는 방법
rabbitmq 메시지 전달 경로
Producer->Switch->Queue->Consumer
일반적으로 3단계로 나누어집니다.
1. 생산자는 메시지 전달의 신뢰성을 보장합니다.
2.mq 내부 메시지는 손실되지 않습니다.
3. 소비자 소비가 성공합니다.
메시지 전달 신뢰성이란 무엇입니까
간단히 말하면 메시지는 100% 메시지 대기열로 전송됩니다.
confirmCallback을 켤 수 있습니다
생산자가 메시지를 전달한 후 mq는 ack를 기반으로 메시지가 mq로 전송되었는지 확인할 수 있습니다.
confirmCallback을 켜세요
수정하세요. 구성 파일
#NONE：禁用发布确认模式，是默认值，CORRELATED：发布消息成功到交换器后会触发回调方法
spring:
rabbitmq:
publisher-confirm-type: correlated테스트 코드
@Test
public void testConfirmCallback() throws InterruptedException {
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
/**
*
* @param correlationData 配置
* @param ack 交换机是否收到消息，true是成功，false是失败
* @param cause 失败的原因
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.out.println("confirm=====>");
System.out.println("confirm==== ack="+ack);
System.out.println("confirm==== cause="+cause);
//根据ACK状态做对应的消息更新操作 TODO
}
});
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"ikun.mei", "鸡你太美");
Thread.sleep(10000);
}는 returnCallback을 통해 Exchange에서 대기열로 메시지가 성공적으로 전송되었는지 확인합니다. 구성 파일 수정
spring:
rabbitmq:
#开启returnCallback
publisher-returns: true
#交换机处理消息到路由失败，则会返回给生产者
template:
mandatory: true테스트 코드
@Test
void testReturnCallback() {
//为true,则交换机处理消息到路由失败，则会返回给生产者 配置文件指定，则这里不需指定
rabbitTemplate.setMandatory(true);
//开启强制消息投递（mandatory为设置为true），但消息未被路由至任何一个queue，则回退一条消息
rabbitTemplate.setReturnsCallback(returned -> {
int code = returned.getReplyCode();
System.out.println("code="+code);
System.out.println("returned="+ returned);
});
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"123456","测试returnCallback");
}소비자가 메시지를 소비할 때 ack를 통해 메시지가 소비되었는지 수동으로 확인해야 합니다.
구성 파일 수정
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual테스트 코드 작성
@RabbitHandler
public void consumer(String body, Message message, Channel channel) throws IOException {
long msgTag = message.getMessageProperties().getDeliveryTag();
System.out.println("msgTag="+msgTag);
System.out.println("message="+ message);
System.out.println("body="+body);
//成功确认，使用此回执方法后，消息会被 rabbitmq broker 删除
channel.basicAck(msgTag,false);
// channel.basicNack(msgTag,false,true);
}deliveryTags는 메시지 전달 순서 번호입니다. 메시지가 소비되거나 메시지가 다시 전달될 때마다 DeliveryTag가 증가합니다
ttl 배달 못한 편지 대기열
데드 레터 큐란 무엇인가요?
has not been 적시에 소비된 메시지가 저장되는 큐
어떤 상황에서 메시지가 데드 레터가 되나요
-
소비자가 메시지를 거부합니다 (basic.reject/ basic. nack) 대기열에 다시 넣지 않습니다. requeue=false
메시지가 대기열에서 소비되지 않았고 대기열 또는 메시지 자체의 만료 시간을 초과했습니다TTL(Time-To-Live)
큐의 메시지 길이가 한계에 도달했습니다
결과: 메시지가 데드 레터가 됩니다. 마지막으로 큐가 데드 레터 스위치에 바인딩된 경우 메시지는 데드 레터 큐로 다시 라우팅됩니다. by the Dead Letter Switch
데드 레터 큐는 지연된 큐 소비를 위해 자주 사용됩니다.
지연 대기열
생산자는 이 메시지가 mq에 전달될 때 즉시 소비될 것으로 기대하지 않지만 소비하기 전에 일정 시간 동안 기다립니다.
springboot는 시간 초과 시 주문 자동 종료를 실현하기 위해 Rabbitmq를 통합합니다.
package com.fandf.test.rabbit;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* @author fandongfeng
* @date 2023/4/15 15:38
*/
@Configuration
public class RabbitMQConfig {
/**
* 订单交换机
*/
public static final String ORDER_EXCHANGE = "order_exchange";
/**
* 订单队列
*/
public static final String ORDER_QUEUE = "order_queue";
/**
* 订单路由key
*/
public static final String ORDER_QUEUE_ROUTING_KEY = "order.#";
/**
* 死信交换机
*/
public static final String ORDER_DEAD_LETTER_EXCHANGE = "order_dead_letter_exchange";
/**
* 死信队列 routingKey
*/
public static final String ORDER_DEAD_LETTER_QUEUE_ROUTING_KEY = "order_dead_letter_queue_routing_key";
/**
* 死信队列
*/
public static final String ORDER_DEAD_LETTER_QUEUE = "order_dead_letter_queue";
/**
* 创建死信交换机
*/
@Bean("orderDeadLetterExchange")
public Exchange orderDeadLetterExchange() {
return new TopicExchange(ORDER_DEAD_LETTER_EXCHANGE, true, false);
}
/**
* 创建死信队列
*/
@Bean("orderDeadLetterQueue")
public Queue orderDeadLetterQueue() {
return QueueBuilder.durable(ORDER_DEAD_LETTER_QUEUE).build();
}
/**
* 绑定死信交换机和死信队列
*/
@Bean("orderDeadLetterBinding")
public Binding orderDeadLetterBinding(@Qualifier("orderDeadLetterQueue") Queue queue, @Qualifier("orderDeadLetterExchange")Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ORDER_DEAD_LETTER_QUEUE_ROUTING_KEY).noargs();
}
/**
* 创建订单交换机
*/
@Bean("orderExchange")
public Exchange orderExchange() {
return new TopicExchange(ORDER_EXCHANGE, true, false);
}
/**
* 创建订单队列
*/
@Bean("orderQueue")
public Queue orderQueue() {
Map<String, Object> args = new HashMap<>(3);
//消息过期后，进入到死信交换机
args.put("x-dead-letter-exchange", ORDER_DEAD_LETTER_EXCHANGE);
//消息过期后，进入到死信交换机的路由key
args.put("x-dead-letter-routing-key", ORDER_DEAD_LETTER_QUEUE_ROUTING_KEY);
//过期时间，单位毫秒
args.put("x-message-ttl", 10000);
return QueueBuilder.durable(ORDER_QUEUE).withArguments(args).build();
}
/**
* 绑定订单交换机和队列
*/
@Bean("orderBinding")
public Binding orderBinding(@Qualifier("orderQueue") Queue queue, @Qualifier("orderExchange")Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(ORDER_QUEUE_ROUTING_KEY).noargs();
}
}Consumer
package com.fandf.test.rabbit;
import cn.hutool.core.date.DateUtil;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* @author fandongfeng
* @date 2023/4/15 15:42
*/
@Component
@RabbitListener(queues = RabbitMQConfig.ORDER_DEAD_LETTER_QUEUE)
public class OrderMQListener {
@RabbitHandler
public void consumer(String body, Message message, Channel channel) throws IOException {
System.out.println("收到消息：" + DateUtil.now());
long msgTag = message.getMessageProperties().getDeliveryTag();
System.out.println("msgTag=" + msgTag);
System.out.println("message=" + message);
System.out.println("body=" + body);
channel.basicAck(msgTag, false);
}
}테스트 클래스
@Test
void testOrder() throws InterruptedException {
//为true,则交换机处理消息到路由失败，则会返回给生产者 配置文件指定，则这里不需指定
rabbitTemplate.setMandatory(true);
//开启强制消息投递（mandatory为设置为true），但消息未被路由至任何一个queue，则回退一条消息
rabbitTemplate.setReturnsCallback(returned -> {
int code = returned.getReplyCode();
System.out.println("code=" + code);
System.out.println("returned=" + returned);
});
rabbitTemplate.convertAndSend(RabbitMQConfig.ORDER_EXCHANGE, "order", "测试订单延迟");
System.out.println("发送消息：" + DateUtil.now());
Thread.sleep(20000);
}프로그램 출력
메시지 보내기: 2023-04-16 15:14:34
메시지 받기: 2023-04 - 16 15:14:44
msgTag=1
message=(본문:'테스트 주문 지연' MessageProperties [headers={spring_listener_return_correlation=03169cfc-5061-41fe-be47-c98e36d17eac, x-first-death-exchange=order_exchange, x - 사망=[{이유=만료, 횟수=1, 교환=주문_교환, 시간=월 4월 16일 15:14:44 CST 2023, 라우팅-키=[주문], 대기열=주문_큐}], x-first-death- 이유 = 만료됨, x-first-death-queue=order_queue}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, 우선순위=0, redelived=false, receivedExchange=order_dead_letter_exchange, receivedRoutingKey=order_dead_letter_queue_routing_key , DeliveryTag=1, ConsumerTag=amq.ctag-Eh8GMgrsrAH1rvtGj7ykOQ, ConsumerQueue=order_dead_letter_queue])
body=테스트 주문 지연
위 내용은 SpringBoot가 RabbitMQ를 통합하여 지연 대기열을 구현하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!