如何避免消息的重复消费问题？（消息消费时的幂等性）

1、 消息的幂等性

https://blog.csdn.net/weixin_63267801/article/details/134211065

1.1、概念

消息的幂等性：就是即使多次收到了消息，也不会重复消费。
对于一个资源，不管你请求一次还是请求多次，对该资源本身造成的影响应该是相同的，不能因为重复的请求而对该资源重复造成影响。

1.2、产生业务场景

同一个消息，第一次接收，正常处理业务，如果该消息第二次再接收，那就不能再处理业务，否则就处理重复了。

2、全局唯一ID+Redis解决消息幂等性问题

2.1、application.yml配置文件

配置rabbitmq和redis

server:port: 8080
spring:application:name: rabbit_13_idempotent01_redisrabbitmq:host: 你的rabbitmq服务器IPport: 5672username: 你的rabbitmq服务管理员账号password: 你的rabbitmq服务管理员密码virtual-host: powerpublisher-confirm-type: correlated #开启交换机的确认模式publisher-returns: truelistener:simple:acknowledge-mode: manual #开启消费者的手动确认模式redis:host: 你的redis服务器IPport: 你的redis服务端口password: 你的redis服务密码database: 1  #1号数据库my:exchangeName: exchange.idempotent.01queueName: queue.idempotent.01

2.2、生产者发送消息

生产者模拟发送两笔相同的订单：
使用com.fasterxml.jackson.databind.ObjectMapper对象进行数据序列化与反序列化：

package com.power.service;import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.power.vo.Orders;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Date;@Service
@Slf4j
public class SendMessage {@Resourceprivate RabbitTemplate rabbitTemplate;//这个对象可以进行序列化和反序列化（json格式）@Resourceprivate ObjectMapper objectMapper;//构造方法执行后执行@PostConstructpublic void init(){//开启生产者的确认模式rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {//如果交换机接收消息成功，ack返回trueif(!ack){log.error("消息没有到达交换机，原因是：{}",cause);//TODO 重发消息或者记录错误日志}});}@Beanpublic void sendMsg() throws JsonProcessingException {{//发送第一笔订单消息Orders orders1 = Orders.builder().orderId("order_100").orderName("手机").money(new BigDecimal("2345")).orderTime(new Date()).build();//将对象转换成jsonlog.info("orders1:::::" + orders1.toString());String strOrders1 = objectMapper.writeValueAsString(orders1);log.info("strOrders1:::::" + strOrders1);MessageProperties messageProperties = new MessageProperties();//设置单条消息持久化，默认技术持久化的messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);Message message = MessageBuilder.withBody(strOrders1.getBytes()).andProperties(messageProperties).build();rabbitTemplate.convertAndSend("exchange.idempotent.01", "info", message);}{//发送第二笔订单消息Orders orders2 = Orders.builder().orderId("order_100").orderName("手机").money(new BigDecimal("2345")).orderTime(new Date()).build();String strOrders2 = objectMapper.writeValueAsString(orders2);MessageProperties messageProperties = new MessageProperties();//设置单条消息持久化，默认技术持久化的messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);Message message = MessageBuilder.withBody(strOrders2.getBytes()).andProperties(messageProperties).build();rabbitTemplate.convertAndSend("exchange.idempotent.01", "info", message);}log.info("消息发送完毕，发送时间是："+new Date());}
}

2.3、消费者接收消息

package com.power.message;import com.fasterxml.jackson.databind.ObjectMapper;
import com.power.vo.Orders;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.io.IOException;@Component
@Slf4j
public class ReceiveMessage {@Resourceprivate ObjectMapper objectMapper;@Resourceprivate StringRedisTemplate stringRedisTemplate;@RabbitListener(queues = {"queue.idempotent.01"})public void receiveMsg(Message message, Channel channel) throws IOException {//获取消息唯一标识long deliveryTag = message.getMessageProperties().getDeliveryTag();//使用objectMapper把字节数组反序列化成对象Orders orders = objectMapper.readValue(message.getBody(), Orders.class);try{log.info("接收到的消息为：{}",orders.toString());//如果不存在就在redis中存储Boolean setResult = stringRedisTemplate.opsForValue().setIfAbsent("idempotent:" + orders.getOrderId(), orders.getOrderId());if(setResult){//TODO 向数据插入订单数据log.info("向数据库插入订单");}//手动确认接收消息成功channel.basicAck(deliveryTag,false);}catch (Exception e){log.error("消息处理出现问题");try {channel.basicNack(deliveryTag,false,true);} catch (IOException ex) {ex.printStackTrace();}throw new RuntimeException(e);}}
}

2.4、pom.xml引入依赖

<?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>com.power</groupId><artifactId>rabbit_13_idempotent01_redis</artifactId><version>1.0-SNAPSHOT</version><name>rabbit_13_idempotent01_redis</name><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.13</version><relativePath/></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.24</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

2.5、RabbitConfig配置类

package com.power.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeName}")private String exchangeName;@Value("${my.queueName}")private String queueName;//创建交换机@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange(exchangeName).build();}//创建队列@Beanpublic Queue queue(){return QueueBuilder.durable(queueName).build();}@Beanpublic Binding binding(DirectExchange directExchange,Queue queue){return BindingBuilder.bind(queue).to(directExchange).with("info");}
}

2.6、启动类

package com.power;import com.power.service.SendMessage;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;import javax.annotation.Resource;@SpringBootApplication
public class Application implements ApplicationRunner {@Resourceprivate SendMessage messageService;public static void main(String[] args) {SpringApplication.run(Application.class);}@Overridepublic void run(ApplicationArguments args) throws Exception {messageService.sendMsg();}
}

2.7、订单对象

package com.power.vo;import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class Orders implements Serializable {private String orderId;private String orderName;private BigDecimal money;private Date orderTime;
}

2.8、测试

启动服务后，生产者发送消息，消费者接收消息，
两笔消息订单ID相同，但是消费者只把接收到的一条消息插入了数据库，实现了消息幂等性。