某某会员小程序后端性能优化

背景

某某会员小程序后台提供开放平台能力,为三方油站提供会员积分、优惠劵等api。当用户在油站加油,油站收银会调用我们系统为用户发放积分、优惠劵等。用户反馈慢,三方调用发放积分接口性能极低,耗时30s+;

接口情况

发放积分接口业务负责,且用存储过程写的业务,改动风险极大

数据库情况

优惠卷等表,数据量800w+,甚至存在单表3000w+

优化方案

数据库数据归档

归档交易、用户优惠劵等表历史数据,比如归档三个月前的数据(根据实际情况补充归档条件,如用户优惠劵没使用或没过期的数据不能归档)
优化效果:存储过程耗时从30s降低到7s,但是作为Toc用途接口性能远远不达标,优化数据库sql或许能进一步降低响应时间,但是存储过程复杂优化费时费力风险大

方案描述风险工作量难度是否能解决性能问题是否解决并发冲突影响使用技术
方案1java重写存储过程业务一定程度能解决yes改动点多,业务影响大java + orm
方案2保证存储过程全局串行执行noyes接口性能会降低分布式锁
方案3异步下存储过程全局串行执行yesyesrabbitmq+分布式锁+自旋锁

线程池异步化分析

接口中存储串行调用改为异步调用,

使用线程池异步化存在问题

开始简单使用线程池异步化,但是出现锁表的情况(原因存储过程没有保证原子性,并且其中大量使用临时表,并发下出现竞争锁表),而SqlServer自带的死锁检查机制杀死事务导致发放积分失败

线程池+分布式锁

异步线程【不能保证分布式环境的全局顺序执行】,使用分布式锁能保证同一个时间只有一个存储过程执行
问题:但是并发情况会将存储过程执行堆积在线程池,并发过大存在OOM风险,或者处理丢失风险

rabbitmq异步改造

可行性验证报告结论

验证通过点如下:

  1. 测试rabbitmq发送/接收消息【通过】
  2. 测试并发下分布式锁+自旋锁保证业务串行执行【通过】
  3. 测试并发下分布式锁+自旋锁+mq保证业务串行执行【通过】
  4. 测试业务幂等性保证不重复消费【通过】
  5. 测试手动ack兼容原来配置保证可靠性【通过】

当前项目rabbitmq使用方式问题分析

配置发下

spring.rabbitmq.host=172.18.229.23
spring.rabbitmq.port=5672
spring.rabbitmq.username=totaltest
spring.rabbitmq.password=totaltest
spring.rabbitmq.virtual-host=/totaltest/
spring.rabbitmq.publisher-confirms=false

该配置没有

spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual

若是不配默认为

spring.rabbitmq.listener.direct.acknowledge-mode=auto
spring.rabbitmq.listener.simple.acknowledge-mode=auto

rabbitmq消费者ack机制问题分析

spring.rabbitmq.listener.direct.acknowledge-mode是用于配置Spring Boot应用中RabbitMQ消息监听器的确认模式。确认模式决定了在消费者处理消息后如何通知RabbitMQ服务器来确认消息的接收情况。
该配置有以下几种可选的值:

  1. AUTO: 在这种模式下,消费者处理消息后,RabbitMQ会自动确认消息。这意味着消息一旦被消费者接收,就会立即从队列中删除。这是默认的确认模式。
  2. MANUAL: 在这种模式下,消费者需要显式地发送确认消息来告知RabbitMQ服务器消息已经被成功处理。这意味着消费者可以在处理消息后决定是否要确认消息。通常在需要进行消息处理的事务性操作时使用这种模式。
  3. NONE: 在这种模式下,消费者不会发送任何确认消息,也不会被要求发送确认消息。这意味着消息会在被传递给消费者之后立即被视为已经被确认。

问题:项目中该配置使用的模式配置,以为着没有手动ack,即消费者接收到消息,消息就会从mq中删除,若是消费者消费异常,则消息丢失不可追溯复原

rabbitmq生产者ack机制问题分析

项目中配置如下

spring.rabbitmq.publisher-confirms=false

spring.rabbitmq.publisher-confirms是Spring Boot中用于配置RabbitMQ生产者消息确认的属性。它用于控制是否启用RabbitMQ的发布确认机制,以确保消息成功发送到Exchange。
当spring.rabbitmq.publisher-confirms属性设置为true时,表示启用了RabbitMQ的发布确认机制。在这种情况下,当生产者发送消息到Exchange后,RabbitMQ会发送一个确认消息给生产者,告知消息是否成功到达Exchange。生产者可以根据收到的确认消息来判断消息是否成功发送,从而进行相应的处理。
当spring.rabbitmq.publisher-confirms属性设置为false时,表示禁用了RabbitMQ的发布确认机制。在这种情况下,生产者发送消息到Exchange后,不会收到确认消息,也无法得知消息是否成功到达Exchange。
通常情况下,建议将spring.rabbitmq.publisher-confirms属性设置为true,以确保消息的可靠发送。当然,具体是否启用发布确认机制,还取决于业务场景和对消息可靠性的要求。

rabbitmq消息可靠性问题分析

通过上诉【rabbitmq生产者ack机制问题分析】和【rabbitmq消费者ack机制问题分析】
可知当前项目中消息没有保证消息可靠性,rabbitmq宕机恢复、消费者消费异常都会导致消息丢失,导致业务完整性缺失

rabbitmq配置最小改动方案

上诉问题若想得到解决需项目中rabbitmq配置,会影响到原来所有使用mq的地方,避免影响范围较大
解决方案:新增消费者类似,通过设置不同的消费者来实现接收指定的消息需要手动 ack

测试rabbitmq配置发送接收消息【通过】

rabbitmq和springboot对应版本:3. Reference
创建虚拟host
image.png
创建测试 交换机和queues

Exchange:exchange-1
Queue:queue-1
key:springboot.*

image.png

spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

image.png

发送消息

package com.bfxy.springboot.producer;import java.util.Map;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;import com.bfxy.springboot.entity.Order;@Component
public class RabbitSender {private static final Logger LOGGER = LoggerFactory.getLogger(RabbitSender.class);//自动注入RabbitTemplate模板类@Autowiredprivate RabbitTemplate rabbitTemplate;  //回调函数: confirm确认final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.err.println("correlationData: " + correlationData);System.err.println("ack: " + ack);if(!ack){System.err.println("异常处理....");}}};//回调函数: return返回final ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,String exchange, String routingKey) {System.err.println("return exchange: " + exchange + ", routingKey: " + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);}};//发送消息方法调用: 构建Message消息public void send(Object message, Map<String, Object> properties)  {LOGGER.info("消息内容:{}",message);LOGGER.info("properties:{}",properties);try {MessageHeaders mhs = new MessageHeaders(properties);Message msg = MessageBuilder.createMessage(message, mhs);rabbitTemplate.setConfirmCallback(confirmCallback);rabbitTemplate.setReturnCallback(returnCallback);//id + 时间戳 全局唯一CorrelationData correlationData = new CorrelationData("1234567890");rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);}catch (Exception e){LOGGER.error("发送消息异常,message:{}",message);}}//发送消息方法调用: 构建自定义对象消息public void sendOrder(Order order)  {LOGGER.info("订单消息内容:{}",order);try {rabbitTemplate.setConfirmCallback(confirmCallback);rabbitTemplate.setReturnCallback(returnCallback);//id + 时间戳 全局唯一CorrelationData correlationData = new CorrelationData("0987654321");rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);}catch (Exception e){LOGGER.error("订单发送消息异常,message:{}",order);}}}

测试代码

package com.bfxy.springboot;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import com.bfxy.springboot.entity.Order;
import com.bfxy.springboot.producer.RabbitSender;@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {@Testpublic void contextLoads() {}@Autowiredprivate RabbitSender rabbitSender;private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");@Testpublic void testSender1() throws Exception {Map<String, Object> properties = new HashMap<>();properties.put("number", "12345");properties.put("send_time", simpleDateFormat.format(new Date()));rabbitSender.send("Hello RabbitMQ For Spring Boot!", properties);}@Testpublic void testSender2() throws Exception {Order order = new Order("001", "第一个订单");rabbitSender.sendOrder(order);}}

接收消息

package com.bfxy.springboot.conusmer;import java.util.Map;import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;import com.rabbitmq.client.Channel;@Component
public class RabbitReceiver {/*** 用于标识方法是一个RabbitMQ消息的监听方法,用于监听指定的队列,并在接收到消息时调用该方法进行处理。* 可以指定队列、交换机、路由键等属性,用于配置消息监听的相关信息。* 通常与@RabbitHandler一起使用,将消息监听和消息处理方法关联起来。*/@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "queue-1",  durable="true"),exchange = @Exchange(value = "exchange-1",  durable="true",  type= "topic",  ignoreDeclarationExceptions = "true"),key = "springboot.*" ))/*** 用于标识方法是一个RabbitMQ消息的处理方法。* 通常与@RabbitListener一起使用,用于指定具体的消息处理方法。* 通过@RabbitHandler注解标识的方法可以处理多个不同类型的消息,通过方法参数的类型来区分不同的消息类型。*/@RabbitHandlerpublic void onMessage(Message message, Channel channel) throws Exception {System.err.println("--------------------------------------");System.err.println("消费端Payload: " + message.getPayload());Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);//手工ACKchannel.basicAck(deliveryTag, false);}/*** * 	spring.rabbitmq.listener.order.queue.name=queue-2spring.rabbitmq.listener.order.queue.durable=truespring.rabbitmq.listener.order.exchange.name=exchange-1spring.rabbitmq.listener.order.exchange.durable=truespring.rabbitmq.listener.order.exchange.type=topicspring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=truespring.rabbitmq.listener.order.key=springboot.** @param order* @param channel* @param headers* @throws Exception*/@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}", durable="${spring.rabbitmq.listener.order.queue.durable}"),exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}", durable="${spring.rabbitmq.listener.order.exchange.durable}", type= "${spring.rabbitmq.listener.order.exchange.type}", ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),key = "${spring.rabbitmq.listener.order.key}"))@RabbitHandlerpublic void onOrderMessage(@Payload com.bfxy.springboot.entity.Order order,  Channel channel,@Headers Map<String, Object> headers) throws Exception {System.err.println("--------------------------------------");System.err.println("消费端order: " + order.getId());Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);//手工ACKchannel.basicAck(deliveryTag, false);}}

断点测试

测试分布式锁+自旋锁测试串行执行【通过】

测试并发分布式锁顺序执行业务代码

package com.bfxy.springboot;import org.junit.Test;
import org.junit.runner.RunWith;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {private static final Logger LOGGER = LoggerFactory.getLogger(ApplicationTests.class);@AutowiredRedissonClient redissonClient;@Testpublic void contextLoads() {long startTime = System.currentTimeMillis();for (int i = 1;i<5;i++){int finalI = i;CompletableFuture.runAsync(()->{bizLock(String.valueOf(finalI));});}while (true){}}private void bizLock(String taskName) {RLock lock = redissonClient.getLock("my-lock");boolean locked = false;try {while (!locked) {locked = lock.tryLock();if (locked) {try {biz(3000, taskName);System.out.println("----------------");} finally {lock.unlock();}} else {// 未获取到锁,可以进行一些等待操作,比如休眠一段时间后再尝试获取锁Thread.sleep(100);}}} catch (Exception e) {e.printStackTrace();}}private void biz(Integer time,String taskName) throws Exception{long startTime = System.currentTimeMillis();LOGGER.info("任务序号={},任务执行开始时间={}",taskName,startTime);Thread.sleep(time);long endtime = System.currentTimeMillis();LOGGER.info("任务序号={},任务执行结束时间={}",taskName,startTime);LOGGER.info("任务序号={},任务执行消耗时间={}",taskName,(endtime-startTime));}
}

执行日志如下:【测试结果并发串行执行(同一时间只有一个任务执行)】

2024-07-10 13:46:49.587  INFO 36076 --- [onPool-worker-9] com.bfxy.springboot.ApplicationTests     : 任务序号=1,任务执行开始时间=1720590409587
2024-07-10 13:46:52.601  INFO 36076 --- [onPool-worker-9] com.bfxy.springboot.ApplicationTests     : 任务序号=1,任务执行结束时间=1720590409587
2024-07-10 13:46:52.601  INFO 36076 --- [onPool-worker-9] com.bfxy.springboot.ApplicationTests     : 任务序号=1,任务执行消耗时间=3014
----------------
2024-07-10 13:46:52.665  INFO 36076 --- [onPool-worker-4] com.bfxy.springboot.ApplicationTests     : 任务序号=4,任务执行开始时间=1720590412665
2024-07-10 13:46:55.678  INFO 36076 --- [onPool-worker-4] com.bfxy.springboot.ApplicationTests     : 任务序号=4,任务执行结束时间=1720590412665
2024-07-10 13:46:55.678  INFO 36076 --- [onPool-worker-4] com.bfxy.springboot.ApplicationTests     : 任务序号=4,任务执行消耗时间=3013
----------------
2024-07-10 13:46:55.759  INFO 36076 --- [onPool-worker-2] com.bfxy.springboot.ApplicationTests     : 任务序号=2,任务执行开始时间=1720590415759
2024-07-10 13:46:58.761  INFO 36076 --- [onPool-worker-2] com.bfxy.springboot.ApplicationTests     : 任务序号=2,任务执行结束时间=1720590415759
2024-07-10 13:46:58.761  INFO 36076 --- [onPool-worker-2] com.bfxy.springboot.ApplicationTests     : 任务序号=2,任务执行消耗时间=3002

压力测试对比资源使用情况

结论使用线程池比较消耗资源,特别是内存,一点并发上来可能oom

压测前

image.png

200并发

image.png

1000并发

image.png

2000并发

image.png

测试分布式锁+自旋锁+mq全局串行执行【通过】

使用线程池控制会导致请求积压到线程池消耗cpu和内存资源,使用mq能有效削峰限流(减小服务器资源消耗),线上部署了两个节点即并发为2

消费者代码

	/*** * 	spring.rabbitmq.listener.order.queue.name=queue-2spring.rabbitmq.listener.order.queue.durable=truespring.rabbitmq.listener.order.exchange.name=exchange-1spring.rabbitmq.listener.order.exchange.durable=truespring.rabbitmq.listener.order.exchange.type=topicspring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=truespring.rabbitmq.listener.order.key=springboot.** @param order* @param channel* @param headers* @throws Exception*/@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}", durable="${spring.rabbitmq.listener.order.queue.durable}"),exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}", durable="${spring.rabbitmq.listener.order.exchange.durable}", type= "${spring.rabbitmq.listener.order.exchange.type}", ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),key = "${spring.rabbitmq.listener.order.key}"))@RabbitHandlerpublic void onOrderMessage(@Payload com.bfxy.springboot.entity.Order order,  Channel channel,@Headers Map<String, Object> headers) throws Exception {//System.err.println("--------------------------------------");//System.err.println("消费端order: " + order.getId());Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);onLockOrderMessage(order);//手工ACKchannel.basicAck(deliveryTag, false);}@AutowiredRedissonClient redissonClient;private void onLockOrderMessage(com.bfxy.springboot.entity.Order order) {RLock lock = redissonClient.getLock("my-lock");boolean locked = false;try {while (!locked) {locked = lock.tryLock();if (locked) {try {long startTime = System.currentTimeMillis();String id = order.getId();LOGGER.info("订单序号={},订单执行开始时间={}",id,startTime);Thread.sleep(7000);long endtime = System.currentTimeMillis();LOGGER.info("订单序号={},订单执行结束时间={}",id,startTime);LOGGER.info("订单序号={},订单执行消耗时间={}",id,(endtime-startTime));System.out.println("----------------");} finally {lock.unlock();}} else {// 未获取到锁,可以进行一些等待操作,比如休眠一段时间后再尝试获取锁Thread.sleep(100);}}} catch (Exception e) {e.printStackTrace();}}

生产者代码

	@Testpublic void testSender3() throws Exception {for (int i = 1;i<=50;i++){int finalI = i;CompletableFuture.runAsync(()->{Order order = new Order(String.valueOf(finalI), "第"+finalI+"个订单");rabbitSender.sendOrder(order);});System.err.println("发送消息订单:"+finalI);if (i%5==0){Thread.sleep(1000);}}}

启动两个消费者【验证全局串行:同一时间只有一个业务执行】

记录消费日志验证是否串行

  1. 通过日志可知:单个消费者消费顺序执行
  2. 验证消费者1和2直接业务串行

消费者2:15:18:22 到 15:18:50 之间没有接收到消息【串行执行】
image.png
验证消费1:15:18:22 到 15:18:50时间段消息情况【串行执行】
image.png

业务幂等性保障测试【通过】

mq接收到消息会将消息中的uid放入redis,当重复消费时会进行判断,保障业务幂等性

	@Testpublic void time() {// 获取字符串对象String key = "myKey";String value = "Hello, Redis!";RBucket<String> bucket = redissonClient.getBucket(key);bucket.set(value, 30, TimeUnit.SECONDS); // 设置失效时间为10秒}

image.png

幂等逻辑

// 判断key是否存在
if(bucket.isExists()){LOGGER.error("重复消费,id={}",id);// 重复消息不执行业务逻辑跳出直接ackbreak;
}else {marker(id);
}

重复消费情况:进入断点表示重复执行break会跳过业务代码
image.png

rabbitmq配置生效测试

原项目配置【自动ack测试】-【通过】

测试自动ack是否生效

package com.bfxy.springboot.config;import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;@Configuration
public class RabbitMQConfig {@Value("${spring.rabbitmq.host}")private String addresses;@Value("${spring.rabbitmq.port}")private String port;@Value("${spring.rabbitmq.username}")private String username;@Value("${spring.rabbitmq.password}")private String password;@Value("${spring.rabbitmq.virtual-host}")private String virtualHost;@Value("${spring.rabbitmq.publisher-confirms}")private boolean publisherConfirms;@Bean/** 因为要设置回调类,所以应是prototype类型,如果是singleton类型,则回调类为最后一次设置 */// @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)public RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());return rabbitTemplate;}@Beanpublic RabbitTemplate manualAckRabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 配置手动ACKrabbitTemplate.setChannelTransacted(true);rabbitTemplate.setChannelTransacted(true);rabbitTemplate.setChannelTransacted(true);rabbitTemplate.setChannelTransacted(true);rabbitTemplate.setAcknowledgeMode(AcknowledgeMode.MANUAL);return rabbitTemplate;}@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setAddresses(addresses + ":" + port);connectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setVirtualHost(virtualHost);/** 如果要进行消息回调,则这里必须要设置为true */connectionFactory.setPublisherConfirms(publisherConfirms);return connectionFactory;}@Bean("mqContainerFactory")@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)public SimpleRabbitListenerContainerFactory containerFactory(ConnectionFactory connectionFactory, MessageConverter messageConverter) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(messageConverter);factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}}

注释手动ack
image.png

spring.rabbitmq.addresses=127.0.0.1:5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtual-host=total-api
spring.rabbitmq.connection-timeout=15000

发送消息,没ack前控制台信息
image.png
等待一会,自动ack的消息从rabbitmq中删除了
image.png

新增配置【手动ack测试】-【通过】

rabbitmq如何实现接受指定的消息要手动ack,其他消息自动ack

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {@Beanpublic RabbitListenerContainerFactory<SimpleMessageListenerContainer> manualAckListenerContainerFactory() {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}@Beanpublic RabbitListenerContainerFactory<SimpleMessageListenerContainer> autoAckListenerContainerFactory() {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setAcknowledgeMode(AcknowledgeMode.AUTO);return factory;}
}
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {@Beanpublic RabbitListenerContainerFactory<SimpleMessageListenerContainer> manualAckListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}@Beanpublic RabbitListenerContainerFactory<SimpleMessageListenerContainer> autoAckListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setAcknowledgeMode(AcknowledgeMode.AUTO);return factory;}
}

消费者代码指定手动ack,并注释手动ackimage.png
查看rabbitmq中消息是否被删除,预期消息不会删除
image.png
放开手动ack注释,再次测试
image.png

兜底保证方案

消息处理可能失败,处理失败的消息记录到broker_message_log表中

-- 表 broker_message_log 消息记录结构
CREATE TABLE IF NOT EXISTS `broker_message_log` (`message_id` varchar(128) NOT NULL, -- 消息唯一ID`message` varchar(4000) DEFAULT NULL, -- 消息内容`try_count` int(4) DEFAULT '0', -- 重试次数`status` varchar(10) DEFAULT '', -- 消息投递状态  0 投递中 1 投递成功   2 投递失败`next_retry` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00',  --下一次重试时间 或 超时时间`create_time` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00', --创建时间`update_time` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00', --更新时间PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

通过定时任务重新执行失败的消息
执行点设计

  1. 定时任务重目标业务方法(该方式要将业务封装某个class的某个方法中,失败时会录入表中)
  2. 发送mq在消费中执行

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/375400.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

VSCode神仙插件——Codeium (AI编程助手)

1、安装&登录插件 安装过程中会让你登录Codeium账户&#xff0c;可以通过Google账户登录&#xff0c;或者可以注册一个Codeium账户&#xff08;如果没有弹出让你登录账户的界面&#xff0c;可以等安装结束后在右下角找到登录的地方&#xff09; 右下角显示如下图所示&#…

Vue3打包发布,刷新出现的空白页面和错误

Vue3打包发布出现的错误&#xff1a;Failed to load module script: Expected a JavaScript module script but the server responded with a MIME type of text/html. Strict MIME type checking is enforced for module scripts per HTML spec. 第一次点击访问到这个路径&…

45 mysql truncate 的实现

前言 truncate 是一个我们也经常会使用到的命令 其作用类似于 delete from $table; 但是 他会比 delete 块很多&#xff0c;这里我们来看一下 它的实现 delete 的时候会逐行进行处理, 打上 删除标记, 然后 由后台任务 进行数据处理 truncate table 的实现 执行 sql 如下 …

C++:从C语言过渡到C++

在这篇博客中&#xff0c;我将会介绍从C语言过渡到C的一些基础知识。 目录 C起源 C的关键字 输出hello&#xff0c;world ​编辑 命名空间 1.什么是命名空间 2.namespace的作用 3.域作用限定符 4.命名空间的使用 IO流 缺省参数 函数重载 引用 1.引用的定义 2.引…

汽车零配件行业看板管理系统应用

生产制造已经走向了精益生产&#xff0c;计算时效产出、物料周转时间等问题&#xff0c;成为每一个制造企业要面临的问题&#xff0c;工厂更需要加快自动化&#xff0c;信息化&#xff0c;数字化的布局和应用。 之前的文章多次讲解了企业MES管理系统&#xff0c;本篇文章就为大…

solidity实战练习1

//SPDX-License-Identifier:MIT pragma solidity ^0.8.24; contract PiggyBank{constructor()payable{emit Deposit(msg.value);//触发事件1//意味着在部署合约的时候&#xff0c;可以向合约发送以太币&#xff08;不是通过调用函数&#xff0c;而是直接在部署合约时发送&#…

中职网络安全wire0077数据包分析

从靶机服务器的FTP上下载wire0077.pcap&#xff0c;分析该文件&#xff0c;找出黑客入侵使用的协议&#xff0c;提交协议名称 SMTP 分析该文件&#xff0c;找出黑客入侵获取的zip压缩包&#xff0c;提交压缩包文件名 DESKTOP-M1JC4XX_2020_09_24_22_43_12.zip 分析该文件&…

26.Labview波形图、XY图、强度图使用精讲

我们如何使用Labview显示曲线或者制作出下面这种我们想要的曲线并随着我们输入值的变化而变化呢&#xff1f; 本文详细讲解一下每种波形图的使用方式&#xff0c;帮助大家深入了解波形图的使用技巧。 文章中所有程序均可在百度网盘下载&#xff0c;下载方式&#xff1a;复制下…

elasticsearch源码分析-04集群状态发布

集群状态发布 cluster模块封装了在集群层面执行的任务&#xff0c;如集群健康、集群级元信息管理、分片分配给节点、节点管理等。集群任务执行之后可能会产生新的集群状态&#xff0c;如果产生新的集群状态主节点会将集群状态广播给其他节点。 集群状态封装在clusterState中&…

python作业二

# 二进制转化为十进制 num input("num:")def binaryToDecimal(binaryString):he 0length len(binaryString)for i in range(length):he int(binaryString[i]) * 2 ** (length - i - 1)return heprint(binaryToDecimal(num))代码运行如下&#xff1a; import math…

ADC 性能规格-静态性能- (2) - 偏移误差( offset error)和满标度增益误差(full scale gain error)

偏移误差(Offset error) 失调(Offset) 定义:失调是指ADC输出数字代码中零位与实际模拟输入零位之间的差异。简单来说,就是当输入信号为零时,ADC输出的数字代码并不一定是零,这个偏差就是失调。影响:失调会影响ADC的整体精度,因为它在整个输入范围内引入了一个固定的偏…

攻防世界 Web_python_template_injection(flask模版注入)

学习文章&#xff1a;https://www.freebuf.com/column/187845.html https://blog.csdn.net/weixin_54515836/article/details/113778233 flask的渲染方法有render_template和render_template_string两种。 render_template()是用来渲染一个指定的文件的。使用如下 return re…

一文读懂DNS和CDN

一.什么是DNS DNS&#xff08;Domain Name System&#xff09;全称为域名系统&#xff0c;是一个将域名和IP地址相互映射的分布式服务&#xff0c;他的作用就是把一个域名解析成为IP地址。我们平时输入的网址&#xff08;或者域名&#xff09;不能被计算机直接识别&#xff0c;…

操作系统——内存管理(面试准备)

虚拟内存 单片机没有操作系统&#xff0c;每次写完代码&#xff0c;都需要借助工具把程序烧录进去&#xff0c;这样程序才能跑起来。 另外&#xff0c;单片机的CPU是直接操作内存的物理地址。 在这种情况下&#xff0c;想在内存中同时运行两个程序是不可能的&#xff0c;如果第…

Python数据分析案例49——基于机器学习的垃圾邮件分类系统构建(朴素贝叶斯,支持向量机)

案例背景 trec06c是非常经典的邮件分类的数据&#xff0c;还是难能可贵的中文数据集。 这个数据集从一堆txt压缩包里面提取出来整理为excel文件还真不容不易&#xff0c;肯定要做一下文本分类。 虽然现在文本分类基本都是深度学习了&#xff0c;但是传统的机器学习也能做。本案…

【论文速读】《面向深度学习的联合消息传递与自编码器》,无线AI的挑战和解决思路

这篇文章来自华为的渥太华无线先进系统能力中心和无线技术实验室&#xff0c;作者中有大名鼎鼎的童文。 一、自编码架构的全局收发机面临的主要问题 文章对我比较有启发的地方&#xff0c;是提到自编码架构的全局收发机面临的主要问题&#xff1a; 问题一&#xff1a;基于随…

【算法笔记自学】第 9 章 提高篇(3)——数据结构专题(2)

9.1树与二叉树 #include <cstdio>int main() {int n, m;scanf("%d%d", &n, &m);printf(n m 1 ? "Yes" : "No");return 0; } 9.2二叉树的遍历 #include <cstdio> #include <vector> using namespace std;const int…

高精度定位与AI技术的深度融合——未来智慧世界的钥匙

引言在当今迅速发展的科技时代&#xff0c;精确定位和人工智能&#xff08;AI&#xff09;技术正在快速推动各领域的创新与变革。高精度定位结合AI技术所产生的融合效应&#xff0c;正在加速智慧城市、智能驾驶、智能物流以及许多其他领域的实现。这篇文章将详细探讨高精度定位…

科技云报道:产业为根大模型应用为擎,容联云推动企业营销服场景重塑

科技云报道原创。 “没有应用&#xff0c;光有一个基础模型&#xff0c;不管是开源还是闭源&#xff0c;一文不值。”在2024世界人工智能大会&#xff08;WAIC 2024&#xff09;现场&#xff0c;百度创始人、董事长兼首席执行官李彦宏直言。 国产大模型的种类越发丰富&#x…

【爬虫】解析爬取的数据

目录 一、正则表达式1、常用元字符2、量词3、Re模块4、爬取豆瓣电影 二、Xpath1、Xpath解析Ⅰ、节点选择Ⅱ、路径表达式Ⅲ、常用函数 2、爬取豆瓣电影 解析数据&#xff0c;除了前面的BeautifulSoup库&#xff0c;还有正则表达式和Xpath两种方法。 一、正则表达式 正则表达式…