backend/java-mall
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复死信消息

Browse Source
This commit is contained in:
Jack 2025-08-29 23:31:26 +08:00
parent de0b4c664a
commit e0f7e56c90
7 changed files with 150 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
mall-account/src/main/java/com/suisung/mall/account/listener/DealUserAnalyticsListener.java
View File

@ -13,6 +13,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.Map;
@Service
@Slf4j
@ -21,16 +22,18 @@ public class DealUserAnalyticsListener {
@Autowired
private AccountUserAnalyticsService accountUserAnalyticsService;
// @RabbitHandler
// public void listener(byte[] data, Channel channel, Message message) throws IOException, InterruptedException {
// // 将byte[]转换为String，然后调用现有的处理逻辑
// String dataStr = new String(data, StandardCharsets.UTF_8);
// listener(dataStr, channel, message);
// }
@RabbitHandler
public void listener(String data, Channel channel, Message message) throws IOException, InterruptedException {
// 检查消息是否已达到最大重试次数
Map<String, Object> headers = message.getMessageProperties().getHeaders();
Object xDeath = headers.get("x-death");
if (xDeath != null) {
log.warn("消息已进入死信队列，可能已达到最大重试次数，直接确认: {}", data);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
return;
}
AccountUserAnalytics accountUserAnalytics = JSONUtil.toBean(data, AccountUserAnalytics.class);
// String messageId = message.getMessageProperties().getMessageId();
@ -40,14 +43,33 @@ public class DealUserAnalyticsListener {
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} else {
log.error("消息消费失败，执行dealUserAnalytics异常，当前用户编号：{}", accountUserAnalytics.getUser_id());
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
// 检查重试次数，避免无限循环
Long retryCount = (Long) headers.getOrDefault("retryCount", 0L);
if (retryCount < 3) {
// 设置重试次数并重新入队
headers.put("retryCount", retryCount + 1);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
} else {
// 达到最大重试次数，直接确认消息
log.error("消息重试次数已达上限，直接确认消息，用户编号：{}", accountUserAnalytics.getUser_id());
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
Thread.sleep(1000);
}
} catch (Exception e) {
log.error("消息消费失败，执行dealUserAnalytics异常，当前用户编号：{}，失败原因", accountUserAnalytics.getUser_id(), e);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
// 检查重试次数，避免无限循环
Long retryCount = (Long) headers.getOrDefault("retryCount", 0L);
if (retryCount < 3) {
// 设置重试次数并重新入队
headers.put("retryCount", retryCount + 1);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
} else {
// 达到最大重试次数，直接确认消息
log.error("消息重试次数已达上限，直接确认消息，用户编号：{}", accountUserAnalytics.getUser_id());
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
Thread.sleep(1000);
}
}
}
}

1
mall-account/src/main/java/com/suisung/mall/account/listener/ExperienceListener.java
View File

@ -13,6 +13,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.Map;
@Service
@Slf4j

80
mall-shop/src/main/java/com/suisung/mall/shop/config/RabbitMqConfig.java
View File

@ -6,8 +6,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
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.SimpleMessageConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@ -22,38 +21,42 @@ public class RabbitMqConfig {
@Autowired
private MqMessageService mqMessageService;
/**
* 序列化
*/
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
// Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
// converter.setAssumeSupportedContentType(true);
// MessageConverter converter = new StringMessageConverter();
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// 消息抵达确认通知
rabbitTemplate.setConfirmCallback((data, ack, cause) -> {
if (ack) {
String msgId = data.getId();
mqMessageService.setMessageStatus(msgId, MqConstant.DELIVERED);
} else {
log.error("消息未能发送成功，消息编号：{}，失败原因：{}", data.getId(), cause);
String msgId = data.getId();
mqMessageService.setMessageStatus(msgId, MqConstant.FAILURE);
}
});
// 消息投递失败通知
rabbitTemplate.setReturnCallback((msg, respCode, respText, exchange, routingKey) -> {
log.error("交换机抵达队列失败，消息编号：{}，状态码：{}，失败原因：{}，当前交换机：{}，当前路由键： {}", msg, respCode, respText, exchange, routingKey);
String msgId = msg.getMessageProperties().getMessageId();
mqMessageService.setMessageStatus(msgId, MqConstant.FAILURE);
});
// 配置序列化配置
rabbitTemplate.setMessageConverter(new SimpleMessageConverter());
return rabbitTemplate;
}
// @Bean
// public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
//
// RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// // 消息抵达确认通知
// rabbitTemplate.setConfirmCallback((data, ack, cause) -> {
// if (ack) {
// String msgId = data.getId();
// mqMessageService.setMessageStatus(msgId, MqConstant.DELIVERED);
// } else {
// log.error("消息未能发送成功，消息编号：{}，失败原因：{}", data.getId(), cause);
//
// String msgId = data.getId();
// mqMessageService.setMessageStatus(msgId, MqConstant.FAILURE);
// }
// });
// // 消息投递失败通知
// rabbitTemplate.setReturnCallback((msg, respCode, respText, exchange, routingKey) -> {
// log.error("交换机抵达队列失败，消息编号：{}，状态码：{}，失败原因：{}，当前交换机：{}，当前路由键： {}", msg, respCode, respText, exchange, routingKey);
// String msgId = msg.getMessageProperties().getMessageId();
// mqMessageService.setMessageStatus(msgId, MqConstant.FAILURE);
// });
//
// // 配置序列化配置
// // rabbitTemplate.setMessageConverter(new SimpleMessageConverter());
//
// Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
// converter.setAssumeSupportedContentType(true);
// rabbitTemplate.setMessageConverter(converter);
//
// return rabbitTemplate;
// }
/**
* 反序列化
@ -62,12 +65,23 @@ public class RabbitMqConfig {
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
// factory.setMessageConverter(new Jackson2JsonMessageConverter());
factory.setMessageConverter(new SimpleMessageConverter());
factory.setMessageConverter(new Jackson2JsonMessageConverter());
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
return factory;
}
/**
* 反序列化
*/
// @Bean
// public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
// SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
// factory.setConnectionFactory(connectionFactory);
// factory.setMessageConverter(new SimpleMessageConverter());
// factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
// return factory;
// }
/**
* 默认交换机
*

20
mall-shop/src/main/java/com/suisung/mall/shop/lakala/controller/mobile/LakalaController.java
View File

@ -93,19 +93,19 @@ public class LakalaController extends BaseControllerImpl {
// List<String> clientIds = JSONUtil.toList(paramsJSON.getJSONArray("clientIds"), String.class);
// return pushMessageService.sendMessage(clientIds, paramsJSON.getStr("title"), paramsJSON.getStr("content"), paramsJSON.getJSONObject("payload"));
// JSONObject jsonObject = new JSONObject();
// String orderId = "DD-20250725-1";
// jsonObject.put("category", 1);
// jsonObject.put("orderId", "DD-20250725-1");
// jsonObject.put("storeId", 12);
// jsonObject.put("title", "有一笔已超时的订单！");
// jsonObject.put("message", "您有一笔已超时的订单[" + orderId + "]，请及时处理。");
// mqMessageService.sendDelayMessage(jsonObject.toString(), 10000);
// return jsonObject;
JSONObject jsonObject = new JSONObject();
String orderId = "DD-20250725-1";
jsonObject.put("category", 1);
jsonObject.put("orderId", "DD-20250725-1");
jsonObject.put("storeId", 12);
jsonObject.put("title", "有一笔已超时的订单！");
jsonObject.put("message", "您有一笔已超时的订单[" + orderId + "]，请及时处理。");
mqMessageService.sendDelayMessage(jsonObject.toString(), 10000);
return jsonObject;
// return shopOrderBaseService.sameCityOrderExpireSeconds(10000L);
return sfExpressApiService.createSfExpressShop(66, "能辉超市", "桂平市", "广西壮族自治区贵港市桂平市广佰汇超市(桂平店)", "谢能坤", "17777525395", "110.07165452271", "23.369069486251");
// return sfExpressApiService.createSfExpressShop(66, "能辉超市", "桂平市", "广西壮族自治区贵港市桂平市广佰汇超市(桂平店)", "谢能坤", "17777525395", "110.07165452271", "23.369069486251");
}

2
mall-shop/src/main/java/com/suisung/mall/shop/lakala/service/impl/LklTkServiceImpl.java
View File

@ -402,7 +402,7 @@ public class LklTkServiceImpl {
String settleType = String.format("D%d", 1);
reqJsonBody.put("settleType", settleType); //结算类型：0-秒到（不分账）；1-次日结算（需要分账）
// formData.put("settlementType", "AUTOMATIC"); // 结算方式：MANUAL:手动结算(结算至拉卡拉APP钱包),AUTOMATIC:自动结算到银行卡,REGULAR:定时结算（仅企业商户支持）
reqJsonBody.put("settlementType", "AUTOMATIC"); // 结算方式：MANUAL:手动结算(结算至拉卡拉APP钱包),AUTOMATIC:自动结算到银行卡,REGULAR:定时结算（仅企业商户支持）
// 店铺省市区信息
Map<String, String> areaCode = getAreaCode(shopMchEntry.getStore_area(), false);

72
mall-shop/src/main/java/com/suisung/mall/shop/order/listener/DelayMessageReceiver.java
View File

@ -41,23 +41,32 @@ public class DelayMessageReceiver {
@Lazy
@Resource
private PushMessageService pushMessageService;
/**
* 监听死信队列，处理过期的延迟消息
*
* @param data 消息内容（JSON格式）
* @param data 消息内容（可能是JSON对象或字符串）
* @param channel RabbitMQ通道，用于手动确认消息
* @param message 消息对象，包含消息属性等信息
*/
@RabbitListener(queues = MqConstant.DEAD_LETTER_QUEUE_NAME)
public void handleExpiredMessage(String data, Channel channel, Message message) {
public void handleExpiredMessage(Object data, Channel channel, Message message) {
log.info("收到过期消息，开始执行触发方法: {}", data);
try {
// 将消息内容转换为字符串格式
String messageStr;
if (data instanceof String) {
messageStr = (String) data;
} else {
// 如果是对象，则转换为JSON字符串
messageStr = new JSONObject(data).toString();
}
// 处理死信消息
boolean success = processDeadMessage(data, channel, message);
boolean success = processDeadMessage(messageStr, channel, message);
if (!success) {
log.warn("处理死信消息失败，消息将重新入队，消息内容: {}", data);
log.warn("处理死信消息失败，消息将重新入队，消息内容: {}", messageStr);
rejectMessage(channel, message); // 处理失败时拒绝消息并重新入队
}
@ -84,11 +93,12 @@ public class DelayMessageReceiver {
}
try {
// 获取消息分类
// 解析消息内容
JSONObject message = new JSONObject(messageStr);
log.debug("解析的消息内容: {}", message);
Integer category = message.getInt("category");
// 获取消息分类
Integer category = message.getInt("category", null);
if (category == null) {
log.warn("消息分类为空，无法处理消息: {}", message);
ackMessage(channel, msg);
@ -101,20 +111,18 @@ public class DelayMessageReceiver {
}
// 根据消息分类处理不同类型的消息
if (category == MqConstant.DEAD_EVENT_CATE_ORDER_EXPIRED) {
// 处理订单超时消息
handleOrderExpiredMessage(message);
return true;
} else if (category == MqConstant.DEAD_EVENT_CATE_PRE_ORDER) {
// 处理预订单消息
handlePreOrderMessage(message);
return true;
if (MqConstant.DEAD_EVENT_CATE_ORDER_EXPIRED.equals(category)) {
log.info("处理订单超时消息: {}", message);
return handleOrderExpiredMessage(message);
} else if (MqConstant.DEAD_EVENT_CATE_PRE_ORDER.equals(category)) {
log.info("处理预订单消息: {}", message);
return handlePreOrderMessage(message);
} else {
log.warn("未知的消息分类: {}，消息内容: {}", category, message);
log.warn("未知的消息分类: {}, 消息内容: {}", category, message);
return false;
}
} catch (Exception e) {
log.error("处理死信消息时发生异常，消息内容: {}, err: {}", messageStr, e);
log.error("处理死信消息时发生异常，消息内容: {}, 错误详情: {}", messageStr, e.getMessage(), e);
return false;
}
}
@ -219,12 +227,20 @@ public class DelayMessageReceiver {
*/
private void ackMessage(Channel channel, Message message) {
try {
// 检查通道是否已关闭
if (!channel.isOpen()) {
log.warn("RabbitMQ通道已关闭，无法确认消息，消息ID: {}", getMessageId(message));
return;
}
// 获取消息ID，优先从标准属性获取，其次从自定义头部获取
String messageId = getMessageId(message);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
log.debug("消息确认成功，消息ID: {}", messageId);
} catch (IOException e) {
log.error("确认消息失败，异常原因：", e);
log.error("确认消息失败，IO异常：", e);
} catch (IllegalStateException e) {
log.warn("确认消息失败，通道已关闭：", e);
}
}
@ -236,16 +252,24 @@ public class DelayMessageReceiver {
*/
private void rejectMessage(Channel channel, Message message) {
try {
// 检查通道是否已关闭
if (!channel.isOpen()) {
log.warn("RabbitMQ通道已关闭，无法拒绝消息，消息ID: {}", getMessageId(message));
return;
}
// 获取消息ID，优先从标准属性获取，其次从自定义头部获取
String messageId = getMessageId(message);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
log.debug("消息拒绝成功，消息将重新入队，消息ID: {}", messageId);
Thread.sleep(1000);
} catch (IOException | InterruptedException e) {
log.error("拒绝消息失败，异常原因：", e);
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
} catch (IOException e) {
log.error("拒绝消息失败，IO异常：", e);
} catch (InterruptedException e) {
log.error("线程中断异常：", e);
Thread.currentThread().interrupt();
} catch (IllegalStateException e) {
log.warn("拒绝消息失败，通道已关闭：", e);
}
}

12
pom.xml
View File

@ -321,12 +321,18 @@
<!-- sentinel配置 -->
<sentinel.transport.dashboard>114.132.210.208:8718</sentinel.transport.dashboard>
<!-- mysql配置 -->
<mysql.host>127.0.0.1</mysql.host>
<mysql.host>114.132.210.208</mysql.host>
<mysql.port>3306</mysql.port>
<mysql.db>mall_dev</mysql.db>
<mysql.user>root</mysql.user>
<mysql.pwd>12345678</mysql.pwd>
<mysql.user>web_dev</mysql.user>
<mysql.pwd>Abc654321$^</mysql.pwd>
<mysql.driver>com.mysql.cj.jdbc.Driver</mysql.driver>
<!-- <mysql.host>127.0.0.1</mysql.host>-->
<!-- <mysql.port>3306</mysql.port>-->
<!-- <mysql.db>mall_dev</mysql.db>-->
<!-- <mysql.user>root</mysql.user>-->
<!-- <mysql.pwd>12345678</mysql.pwd>-->
<!-- <mysql.driver>com.mysql.cj.jdbc.Driver</mysql.driver>-->
<!-- redis配置 -->
<redis.host>114.132.210.208</redis.host>
<redis.database>15</redis.database>