一、创建消费者

二、创建订单链路配置

1.定义RabbitMQ配置类

2.创建RabbitmqOrderConfig配置类

三、如何实现RabbitMQ重复投递机制

1.开启发送者消息确认模式

2.消息发送确认

① 创建ConfirmCallBacker确认模式

② 创建ReturnCallBack退回模式

3.创建生产者

4.创建消费者（手动消费）

5.启动测试

6.采坑日记

① 异常点一：@RabbitListener

② 异常点二：手动确认消息

③ 异常点三：消息格式

④ 异常点四：消息不确认

⑤ 异常点五：消息无限投递

⑥ 异常点六：重复消费

四、RabbitMQ在项目中的秒杀业务优化

1.修改秒杀订单生产方式

2.消费者监听器完成秒杀订单生成

一、创建消费者

第1步：基于Spring Initialzr方式创建zmall-rabbitmq消费者模块

第2步：在公共模块中添加rabbitmq相关依赖

<!--rabbitmq-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

第3步：配置子模块zmall-rabbitmq的pom.xml，引入公共模块zmall-common

<dependencies><dependency><groupId>com.zking.zmall</groupId><artifactId>zmall-common</artifactId><version>1.0-SNAPSHOT</version></dependency>
</dependencies>

第4步：配置父模块的pom.xml，添加子模块zmall-rabbitmq

<modules><module>zmall-common</module>...<module>zmall-rabbitmq</module>
</modules>

第5步：配置application.yml

server:port: 8050
spring:application:name: zmall-rabbitmqdatasource:#type连接池类型 DBCP,C3P0,Hikari,Druid,默认为Hikaritype: com.zaxxer.hikari.HikariDataSourcedriver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://localhost:3306/zmall?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=trueusername: rootpassword: 123456cloud:nacos:config:server-addr: localhost:8848redis:host: localhostport: 6379password: 123456jedis:pool:max-active: 100max-wait: 10max-idle: 10min-idle: 10database: 0rabbitmq:host: 192.168.29.128port: 5672username: adminpassword: adminvirtual-host: my_vhost# 发送者开启 confirm 确认机制#publisher-confirm-type: correlated# 发送者开启 return 确认机制#publisher-returns: true# 设置消费端手动 acklistener:simple:#手动应答acknowledge-mode: manual#消费端最小并发数concurrency: 5#消费端最大并发数max-concurrency: 10#一次请求中预处理的消息数量prefetch: 5# 是否支持重试retry:#启用消费重试enabled: true#重试次数max-attempts: 3#重试间隔时间initial-interval: 3000cache:channel:#缓存的channel数量size: 50
#mybatis-plus配置
mybatis-plus:#所对应的 XML 文件位置mapper-locations: classpath*:/mapper/*Mapper.xml#别名包扫描路径type-aliases-package: com.zking.zmall.modelconfiguration:#驼峰命名规则map-underscore-to-camel-case: true
#日志配置
logging:level:com.zking.zmall.mapper: debug

注意：这里面的配置，记得要改换为你的虚拟机ID

消费者采用的是手动消费模式，请注意设置

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

第6步：配置启动类

package com.zking.zmallrabbitmq;import org.checkerframework.checker.units.qual.Mass;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;@EnableFeignClients
@EnableDiscoveryClient
@MapperScan({"com.zking.zmall.mapper","com.zking.zmallrabbitmq.mapper"})
@SpringBootApplication
public class ZmallRabbitmqApplication {public static void main(String[] args) {SpringApplication.run(ZmallRabbitmqApplication.class, args);}}

启动项目的时候，记得打开我们的nacos注册中心，测试一下看看有没有问题。能运行起来说明就没有问题。

二、创建订单链路配置

1.定义RabbitMQ配置类

定义RabbitMQ配置类，设置生成者发送数据时自动转换成JSON，设置消费者获取消息自动转换成JSON。

package com.zking.zmallrabbitmq.config;import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqConfig {@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate template = new RabbitTemplate(connectionFactory);template.setMessageConverter(new Jackson2JsonMessageConverter());return template;}@Beanpublic SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setMessageConverter(new Jackson2JsonMessageConverter());return factory;}
}

设置RabbitTemplate消息转换模式为Jackson2JsonMessageConverter；

设置RabbitMQ消费者监听器的的消息转换模式为Jackson2JsonMessageConverter；

2.创建RabbitmqOrderConfig配置类

创建RabbitmqOrderConfig配置类，增加订单队列、交换机及绑定关系。

package com.zking.zmallrabbitmq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqOrderConfig {public static final String ORDER_QUEUE="order-queue";public static final String ORDER_EXCHANGE="order-exchange";public static final String ORDER_ROUTING_KEY="order-routing-key";@Beanpublic Queue orderQueue(){return new Queue(ORDER_QUEUE,true);}@Beanpublic DirectExchange orderExchange(){return new DirectExchange(ORDER_EXCHANGE,true,false);}@Beanpublic Binding orderBinding(){return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ORDER_ROUTING_KEY);}
}

三、如何实现RabbitMQ重复投递机制

1.开启发送者消息确认模式

配置application.yml，开启发送者confirm确认机制和return确认机制

spring:rabbitmq:# 发送者开启 confirm 确认机制publisher-confirm-type: correlated# 发送者开启 return 确认机制publisher-returns: true

2.消息发送确认

rabbitmq 的消息确认分为两部分：发送消息确认和消息接收确认

发送消息确认：用来确认生产者 producer 将消息发送到 broker ，broker 上的交换机 exchange 再投递给队列 queue的过程中，消息是否成功投递。

消息从 producer 到 rabbitmq broker有一个 confirmCallback 确认模式。

消息从 exchange 到 queue 投递失败有一个 returnCallback 退回模式。

我们可以利用这两个Callback来确保消息100%送达。

Broker：简单来说,就是一个消息队列服务器实体。

① 创建ConfirmCallBacker确认模式

package com.zking.zmallrabbitmq.component;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class ConfirmCallbackService implements RabbitTemplate.ConfirmCallback {/**** @param correlationData 对象内部只有一个 id 属性，用来表示当前消息的唯一性* @param ack 消息投递到broker 的状态，true表示成功* @param cause 表示投递失败的原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (!ack) {log.error("消息发送异常!");} else {log.info("发送者已经收到确认，ack={}, cause={}",ack, cause);}}
}

② 创建ReturnCallBack退回模式

@Slf4j
@Component
public class ReturnCallbackService implements RabbitTemplate.ReturnCallback {/**** @param message 消息体* @param replyCode 响应code* @param replyText 响应内容* @param exchange 交换机* @param routingKey 路由键*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.info("returnedMessage ===> replyCode={} ,replyText={} ,exchange={} ,routingKey={}", replyCode, replyText, exchange, routingKey);}
}

3.创建生产者

创建生产者，模拟发送消息

package com.zking.zmallrabbitmq.controller;import com.zking.zmall.model.Order;
import com.zking.zmallrabbitmq.component.ConfirmCallbackService;
import com.zking.zmallrabbitmq.component.ReturnCallbackService;
import com.zking.zmallrabbitmq.config.RabbitmqOrderConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.Date;@RestController
public class ProducerController {@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate ReturnCallbackService returnCallbackService;@Autowiredprivate ConfirmCallbackService confirmCallbackService;@RequestMapping("/sendMessage")public void sendMessage(){Order order=new Order();order.setId(1);order.setUserId(2);order.setLoginName("zhangsan");order.setUserAddress("长沙");order.setCreateTime(new Date());order.setCost(120.0F);order.setSerialNumber("123");order.setState(0);//ConfirmCallback确认模式rabbitTemplate.setConfirmCallback(confirmCallbackService);//ReturnCallback退回模式rabbitTemplate.setReturnCallback(returnCallbackService);rabbitTemplate.convertAndSend(RabbitmqOrderConfig.ORDER_EXCHANGE,RabbitmqOrderConfig.ORDER_ROUTING_KEY,order);}
}

4.创建消费者（手动消费）

package com.zking.zmallrabbitmq.component;import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.zking.zmall.model.Order;
import lombok.extern.slf4j.Slf4j;
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;
import java.util.HashMap;
import java.util.Map;@Slf4j
@Component
public class OrderConsumerListener {//最大重试次数private static final Integer MAX_RECONSUME_COUNT=3;//用于记录消息重试次数的集合，可以采用Redis方式实现private static Map<String,Integer> retryMap=new HashMap<>();@RabbitHandler@RabbitListener(queues = {"order-queue"},ackMode = "MANUAL")public void recieveMessage(Message message,Order order,Channel channel) throws IOException {//channel内按顺序自增的消息IDlong deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("接收到消息:"+message+",消息内容:"+ JSON.toJSONString(order));//模拟异常，开始消息重试int i= 1/0;} catch (Exception e) {e.printStackTrace();String msgId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");Integer retryCount = retryMap.get(msgId)==null?1:retryMap.get(msgId);log.info("即将开始第{}次消息重试....",retryCount);if(retryCount>=MAX_RECONSUME_COUNT){log.info("重试次数达到3次，消息被拒绝，retryCount="+retryCount);//此处要注意：当重试次数到达3次后，将拒绝消息且不在重新入队列channel.basicReject(deliverTag,false);}else{//重新发送消息到队尾//再次发送该消息到消息队列，异常消息就放在了消息队列尾部，这样既保证消息不会丢失，又保证了正常业务的进行。channel.basicPublish(message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.MINIMAL_PERSISTENT_BASIC,JSON.toJSONBytes(order));}retryMap.put(msgId,retryCount+1);}//成功确认消息，非批量模式channel.basicAck(deliverTag, false);}
}

5.启动测试

访问：http://localhost:8050/sendMessage

6.采坑日记

① 异常点一：@RabbitListener

异常原因：@RabbitListener作用于类上引发异常；解决方案：@RabbitListener移至消费者监听器的方法上，而@RabbitListener只适用于方法级别。

② 异常点二：手动确认消息

虽然在消费者端的application.yml中配置手动消费模式，但是在服务消费时引发了这个异常错误，导致重复消费的问题。原因是使用@RabbitListener注解会自动ACK，如果方法中再手动ACK会造成重复ACK，所以报错；解决方式就是在@RabbitListener中配置手动消费模式：ackMode = "MANUAL"。

③ 异常点三：消息格式

在消费者消费消息时引发异常，触发消息重新投递，但是由于重新投递时导致消息格式问题引发了消息转换异常。具体原因通过查看日志发现，重新投递的消息格式为text/plain，而我们在处理消息时采用的是json方式，导致消息转换异常。解决方案：将重新发送消息的状态由MessageProperties.PERSISTENT_TEXT_PLAIN更改为MessageProperties.MINIMAL_PERSISTENT_BASIC

④ 异常点四：消息不确认

这是一个非常没技术含量的坑，但却是非常容易犯错的地方。开启消息确认机制，消费消息别忘了channel.basicAck，否则消息会一直存在，导致重复消费。

⑤ 异常点五：消息无限投递

最开始接触消息确认机制的时候，消费端代码就像下边这样写的，思路很简单：处理完业务逻辑后确认消息， int a = 1 / 0 发生异常后将消息重新投入队列

@RabbitHandler
public void recieveMessage(Message message,Order order,Channel channel) throws IOException {//channel内按顺序自增的消息IDlong deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("接收到消息:"+message+",消息内容:"+ JSON.toJSONString(order));int i = 1 / 0;channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}
}

但是有个问题是，业务代码一旦出现 bug 99.9%的情况是不会自动修复，一条消息会被无限投递进队列，消费端无限执行，导致了死循环。

经过测试分析发现，当消息重新投递到消息队列时，这条消息不会回到队列尾部，仍是在队列头部。

消费者会立刻消费这条消息，业务处理再抛出异常，消息再重新入队，如此反复进行。导致消息队列处理出现阻塞，导致正常消息也无法运行。

而解决方案是，先将消息进行应答，此时消息队列会删除该条消息，同时我们再次发送该消息到消息队列，异常消息就放在了消息队列尾部，这样既保证消息不会丢失，又保证了正常业务的进行。

//重新发送消息到队尾
//再次发送该消息到消息队列，异常消息就放在了消息队列尾部，这样既保证消息不会丢失，又保证了正常业务的进行。
channel.basicPublish(message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(), 								 MessageProperties.MINIMAL_PERSISTENT_BASIC,JSON.toJSONBytes(order));

⑥ 异常点六：重复消费

如何保证 MQ 的消费是幂等性，这个需要根据具体业务而定，可以借助MySQL、或者redis 将消息持久化。

上述代码结构图

四、RabbitMQ在项目中的秒杀业务优化

1.修改秒杀订单生产方式

第1步：修改zmall-order订单模块的application.yml，加入rabbitmq相关配置

spring:rabbitmq:host: 192.168.195.143port: 5672username: adminpassword: adminvirtual-host: my_vhost# 设置消费端手动 acklistener:simple:acknowledge-mode: manual# 是否支持重试retry:enabled: truemax-attempts: 3

第2步：修改秒杀订单生成方式，针对抢购成功的秒杀订单直接推送到RabbitMQ中

OrderServiceImpl

@Autowiredprivate RabbitTemplate rabbitTemplate;@Transactional@Overridepublic JsonResponseBody<?> createKillOrder(User user, Integer pid, Float price) {//判断用户是否登录if(null==user)throw new BusinessException(JsonResponseStatus.TOKEN_ERROR);//根据秒杀商品ID和用户ID判断是否重复抢购Order order = redisService.getKillOrderByUidAndPid(user.getId(), pid);if(null!=order)return new JsonResponseBody<>(JsonResponseStatus.ORDER_REPART);//Redis库存预减long stock = redisService.decrement(pid);if(stock<0){redisService.increment(pid);return new JsonResponseBody<>(JsonResponseStatus.STOCK_EMPTY);}//创建订单order=new Order();order.setUserId(user.getId());order.setLoginName(user.getLoginName());order.setPid(pid);order.setCost(price);//将生成的秒杀订单保存到Redis中redisService.setKillOrderToRedis(pid,order);//将生成的秒杀订单推送到RabbitMQ中的订单队列中rabbitTemplate.convertAndSend(RabbitmqOrderConfig.ORDER_EXCHANGE,RabbitmqOrderConfig.ORDER_ROUTING_KEY,order);return new JsonResponseBody<>();}

Order

//    数据库不存在的字段 
@TableField(exist =  false)private  Integer pid;

2.消费者监听器完成秒杀订单生成

第1步：将zmall-order订单模块中的service业务处理接口及实现类移至消息者监听器模块。

第2步：zmall-rabbitmq模块中在IOrderService及OrderServiceImpl中重新定义生成秒杀订单方法

public interface IOrderService extends IService<Order> {void saveOrder(Order order);
}

package com.zking.zmallrabbitmq.service.impl;import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zking.zmall.config.RabbitmqOrderConfig;
import com.zking.zmall.mapper.OrderMapper;
import com.zking.zmall.model.Kill;
import com.zking.zmall.model.Order;
import com.zking.zmall.model.OrderDetail;
import com.zking.zmall.model.User;
import com.zking.zmall.service.impl.RedisServiceImpl;
import com.zking.zmall.util.JsonResponseBody;
import com.zking.zmall.util.JsonResponseStatus;
import com.zking.zmall.util.SnowFlake;
import com.zking.zmallrabbitmq.service.IOrderService;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;//import io.seata.spring.annotation.GlobalTransactional;/*** <p>*  服务实现类* </p>** @author zking* @since 2023-02-08*/
@Service
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements IOrderService {@Autowiredprivate KillServiceImpl killService;@Autowiredprivate OrderDetailServiceImpl orderDetailService;@Autowiredprivate RabbitTemplate rabbitTemplate;@Transactional@Overridepublic void saveOrder(Order order) {//2.秒杀商品库存减一boolean flag=killService.update(new UpdateWrapper<Kill>().setSql("total=total-1").eq("item_id",order.getPid()).gt("total",0));if(!flag)return;//3.生成秒杀订单及订单项SnowFlake snowFlake=new SnowFlake(2,3);Long orderId=snowFlake.nextId();int orderIdInt = new Long(orderId).intValue();//订单order.setSerialNumber(orderIdInt+"");this.save(order);//订单项OrderDetail orderDetail=new OrderDetail();orderDetail.setOrderId(orderIdInt);orderDetail.setProductId(order.getPid());orderDetail.setQuantity(1);orderDetail.setCost(order.getCost());orderDetailService.save(orderDetail);}}

第3步：修改秒杀订单消费者监听器

OrderDetailServiceImpl

package com.zking.zmallrabbitmq.component;import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.zking.zmall.model.Order;
import com.zking.zmallrabbitmq.service.IOrderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;@Slf4j
@Component
public class OrderConsumerListener {@Autowiredprivate IOrderService orderService;//最大重试次数private static final Integer MAX_RECONSUME_COUNT=3;//用于记录消息重试次数的集合，可以采用Redis方式实现private static Map<String,Integer> retryMap=new HashMap<>();@RabbitHandler@RabbitListener(queues = {"order-queue"},ackMode = "MANUAL")public void recieveMessage(Message message,Order order,Channel channel) throws IOException {//channel内按顺序自增的消息IDlong deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("接收到消息:"+message+",消息内容:"+ JSON.toJSONString(order));//模拟异常，开始消息重试
//            int i= 1/0;
//            修改库存、生成订单表数据，生成订单项表数据
//            保存秒杀订单及订单项orderService.saveOrder(order);} catch (Exception e) {e.printStackTrace();String msgId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");Integer retryCount = retryMap.get(msgId)==null?1:retryMap.get(msgId);log.info("即将开始第{}次消息重试....",retryCount);if(retryCount>=MAX_RECONSUME_COUNT){log.info("重试次数达到3次，消息被拒绝，retryCount="+retryCount);//此处要注意：当重试次数到达3次后，将拒绝消息且不在重新入队列channel.basicReject(deliverTag,false);}else{//重新发送消息到队尾//再次发送该消息到消息队列，异常消息就放在了消息队列尾部，这样既保证消息不会丢失，又保证了正常业务的进行。channel.basicPublish(message.getMessageProperties().getReceivedExchange(),message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.MINIMAL_PERSISTENT_BASIC,JSON.toJSONBytes(order));}retryMap.put(msgId,retryCount+1);}//成功确认消息，非批量模式channel.basicAck(deliverTag, false);}
}