Browse Source

bug处理

feature-1.1
wuxicheng 3 years ago
parent
commit
af3f10ace2
  1. 69
      bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/template/RocketMQEnhanceTemplate.java
  2. 6
      bnyer-services/bnyer-order/src/main/java/com/bnyer/order/listener/vip/VipOrderPayNotifyConsumer.java
  3. 2
      bnyer-services/bnyer-pay/src/main/java/com/bnyer/pay/service/impl/PayInfoServiceImpl.java

69
bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/template/RocketMQEnhanceTemplate.java

@ -27,16 +27,6 @@ import javax.annotation.Resource;
@RequiredArgsConstructor(onConstructor = @__(@Autowired)) @RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class RocketMQEnhanceTemplate { public class RocketMQEnhanceTemplate {
/**
* 延时等级
*/
private static final int DELAY_LEVEL = RocketMqConstant.FIVE_SECOND;
/**
* 默认重试次数
*/
private static final int MAX_RETRY_TIMES = 3;
private final RocketMQTemplate template; private final RocketMQTemplate template;
@Resource @Resource
@ -152,14 +142,6 @@ public class RocketMQEnhanceTemplate {
}); });
} }
/**
* 发送异步可重试消息通过线程池执行发送到broker的消息任务执行完后回调如果发送失败进行重试
*
*/
public <T extends BaseMessage> void sendAsyncRetryMsg(String topic, String tag, T message) {
sendAsyncMsg(buildDestination(topic,tag),message,getDelayLevel());
}
/** /**
* 发送异步延时消息通过线程池执行发送到broker的消息任务执行完后回调在SendCallback中可处理相关成功失败时的逻辑 * 发送异步延时消息通过线程池执行发送到broker的消息任务执行完后回调在SendCallback中可处理相关成功失败时的逻辑
* *
@ -182,57 +164,8 @@ public class RocketMQEnhanceTemplate {
@Override @Override
public void onException(Throwable throwable) { public void onException(Throwable throwable) {
log.error("消息发送失败,destination:{},error:{}",destination,throwable.getMessage()); log.error("消息发送失败,destination:{},error:{}",destination,throwable.getMessage());
// 超过最大重试次数时调用子类方法处理
if (message.getRetryTimes() > getMaxRetryTimes()) {
handleMaxRetriesExceeded(message);
return;
}
//发送失败,重试发送
handleRetry(destination,message);
}
/**
* 重试发送异步消息
* @param destination
* @param message
* @param <T>
*/
private <T extends BaseMessage> void handleRetry(String destination,T message) {
//重新构建消息体
String messageSource = message.getSource();
if(!messageSource.startsWith(RocketMqConstant.RETRY_PREFIX)){
message.setSource(RocketMqConstant.RETRY_PREFIX + messageSource);
}
message.setRetryTimes(message.getRetryTimes() + 1);
// 如果消息发送不成功,则再次重新发送
sendAsyncMsg(destination, message,getDelayLevel());
} }
},10000,delayLevel);
/**
* 超过最大重试次数调用该方法处理
* @param message
* @param <T>
*/
private <T extends BaseMessage> void handleMaxRetriesExceeded(T message) {
log.error("发送消息超过最大重试次数,message:{}",message);
}
},delayLevel);
}
/**
* isRetry开启时重新入队延迟时间
* @return -1立即入队重试
*/
protected int getDelayLevel() {
return DELAY_LEVEL;
}
/**
* 最大重试次数
*
* @return 最大重试次数默认3次
*/
protected int getMaxRetryTimes() {
return MAX_RETRY_TIMES;
} }
} }

6
bnyer-services/bnyer-order/src/main/java/com/bnyer/order/listener/vip/VipOrderPayNotifyConsumer.java

@ -55,11 +55,11 @@ public class VipOrderPayNotifyConsumer extends EnhanceMessageHandler<VipOrderPay
vipOrderService.updateByToPaySuccess(vipOrder); vipOrderService.updateByToPaySuccess(vipOrder);
//发消息,添加用户会员记录 //发消息,添加用户会员记录
AddUserVipRecordMessage addUserVipRecordMessage = buildVipRecordMsg(vipOrder); AddUserVipRecordMessage addUserVipRecordMessage = buildVipRecordMsg(vipOrder);
rocketMQEnhanceTemplate.sendAsyncRetryMsg(RocketMqTopic.VIP_RECORD_CREATE_TOPIC,null, addUserVipRecordMessage); rocketMQEnhanceTemplate.sendAsyncMsg(RocketMqTopic.VIP_RECORD_CREATE_TOPIC,null, addUserVipRecordMessage);
//发送开会员画意值奖励并写入记录消息 //发送开会员画意值奖励并写入记录消息
GoldRewardMessage goldMsg = buildGoldRewardMsg(vipOrder.getUserId(),300, GoldEnum.BUY_VIP.getCode(),null,vipOrder.getUserClientType()); GoldRewardMessage goldMsg = buildGoldRewardMsg(vipOrder.getUserId(),300, GoldEnum.BUY_VIP.getCode(),null,vipOrder.getUserClientType());
rocketMQEnhanceTemplate.sendAsyncRetryMsg(RocketMqTopic.GOLD_REWARD_TOPIC,null,goldMsg); rocketMQEnhanceTemplate.sendAsyncMsg(RocketMqTopic.GOLD_REWARD_TOPIC,null,goldMsg);
} }
@ -119,7 +119,7 @@ public class VipOrderPayNotifyConsumer extends EnhanceMessageHandler<VipOrderPay
@Override @Override
protected boolean filter(VipOrderPayNotifyMessage message) { protected boolean filter(VipOrderPayNotifyMessage message) {
return true; return false;
} }
@Override @Override

2
bnyer-services/bnyer-pay/src/main/java/com/bnyer/pay/service/impl/PayInfoServiceImpl.java

@ -86,7 +86,7 @@ public class PayInfoServiceImpl extends ServiceImpl<PayInfoMapper, PayInfo> impl
vipOrderPayNotifyMessage.setKey(IdUtils.randomUUID()); vipOrderPayNotifyMessage.setKey(IdUtils.randomUUID());
vipOrderPayNotifyMessage.setSource(ServiceNameConstants.PAY_SERVICE); vipOrderPayNotifyMessage.setSource(ServiceNameConstants.PAY_SERVICE);
vipOrderPayNotifyMessage.setOrderNo(orderNo); vipOrderPayNotifyMessage.setOrderNo(orderNo);
rocketMQEnhanceTemplate.sendAsyncRetryMsg(RocketMqTopic.VIP_ORDER_PAY_NOTIFY_TOPIC,null, vipOrderPayNotifyMessage); rocketMQEnhanceTemplate.sendAsyncMsg(RocketMqTopic.VIP_ORDER_PAY_NOTIFY_TOPIC,null, vipOrderPayNotifyMessage);
break; break;
} }
} }

Loading…
Cancel
Save