caojiahao/dev_caojiahao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

!199 优化监听事务提交发送mq事件，为订单变更发送mq增加事务提交监听

Browse Source 
Merge pull request !199 from OceansDeep/feature/pg
This commit is contained in:
OceansDeep 2022-06-22 09:46:45 +00:00 committed by Gitee
commit 5912428c00
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
10 changed files with 101 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
framework/src/main/java/cn/lili/common/event/TransactionCommitSendMQEvent.java Normal file
View File

@ -0,0 +1,32 @@
package cn.lili.common.event;
import lombok.Getter;
import org.springframework.context.ApplicationEvent;
/**
* 事务提交后发生mq事件
*
* @author paulG
* @since 2022/1/19
**/
public class TransactionCommitSendMQEvent extends ApplicationEvent {
private static final long serialVersionUID = 5885956821347953071L;
@Getter
private final String topic;
@Getter
private final String tag;
@Getter
private final String message;
public TransactionCommitSendMQEvent(Object source, String topic, String tag, String message) {
super(source);
this.topic = topic;
this.tag = tag;
this.message = message;
}
}

38
framework/src/main/java/cn/lili/common/listener/TransactionCommitSendMQListener.java Normal file
View File

@ -0,0 +1,38 @@
package cn.lili.common.listener;
import cn.lili.common.event.TransactionCommitSendMQEvent;
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
/**
* 事务提交监听器
*
* @author paulG
* @since 2022/1/19
**/
@Component
@Slf4j
public class TransactionCommitSendMQListener {
/**
* rocketMq
*/
@Autowired
private RocketMQTemplate rocketMQTemplate;
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void send(TransactionCommitSendMQEvent event) {
log.info("事务提交，发送mq信息!{}", event);
String destination = event.getTopic() + ":" + event.getTag();
//发送订单变更mq消息
rocketMQTemplate.asyncSend(destination, event.getMessage(), RocketmqSendCallbackBuilder.commonCallback());
}
}

24
framework/src/main/java/cn/lili/modules/goods/event/GeneratorEsGoodsIndexEvent.java
View File

@ -1,24 +0,0 @@
package cn.lili.modules.goods.event;
import lombok.Data;
import org.springframework.context.ApplicationEvent;
/**
* @author paulG
* @since 2022/1/19
**/
@Data
public class GeneratorEsGoodsIndexEvent extends ApplicationEvent {
private static final long serialVersionUID = -6206752641309458207L;
private String id;
private String tag;
public GeneratorEsGoodsIndexEvent(Object source, String tag, String id) {
super(source);
this.tag = tag;
this.id = id;
}
}

37
framework/src/main/java/cn/lili/modules/goods/listener/GeneratorEsGoodsIndexListener.java
View File

@ -1,37 +0,0 @@
package cn.lili.modules.goods.listener;
import cn.lili.common.properties.RocketmqCustomProperties;
import cn.lili.modules.goods.event.GeneratorEsGoodsIndexEvent;
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
/**
* @author paulG
* @since 2022/1/19
**/
@Component
public class GeneratorEsGoodsIndexListener {
/**
* rocketMq
*/
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* rocketMq配置
*/
@Autowired
private RocketmqCustomProperties rocketmqCustomProperties;
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void generatorEsGoodsIndex(GeneratorEsGoodsIndexEvent esGoodsIndexEvent) {
String destination = rocketmqCustomProperties.getGoodsTopic() + ":" + esGoodsIndexEvent.getTag();
//发送mq消息
rocketMQTemplate.asyncSend(destination, esGoodsIndexEvent.getId(), RocketmqSendCallbackBuilder.commonCallback());
}
}

8
framework/src/main/java/cn/lili/modules/goods/serviceimpl/GoodsSkuServiceImpl.java
View File

@ -9,6 +9,7 @@ import cn.lili.cache.Cache;
import cn.lili.cache.CachePrefix;
import cn.lili.common.enums.PromotionTypeEnum;
import cn.lili.common.enums.ResultCode;
import cn.lili.common.event.TransactionCommitSendMQEvent;
import cn.lili.common.exception.ServiceException;
import cn.lili.common.properties.RocketmqCustomProperties;
import cn.lili.common.security.context.UserContext;
@ -26,7 +27,6 @@ import cn.lili.modules.goods.entity.vos.GoodsSkuSpecVO;
import cn.lili.modules.goods.entity.vos.GoodsSkuVO;
import cn.lili.modules.goods.entity.vos.GoodsVO;
import cn.lili.modules.goods.entity.vos.SpecValueVO;
import cn.lili.modules.goods.event.GeneratorEsGoodsIndexEvent;
import cn.lili.modules.goods.mapper.GoodsSkuMapper;
import cn.lili.modules.goods.service.*;
import cn.lili.modules.goods.sku.GoodsSkuBuilder;
@ -362,10 +362,10 @@ public class GoodsSkuServiceImpl extends ServiceImpl<GoodsSkuMapper, GoodsSku> i
boolean update = this.update(updateWrapper);
if (Boolean.TRUE.equals(update)) {
if (GoodsStatusEnum.UPPER.name().equals(marketEnable)) {
applicationEventPublisher.publishEvent(new GeneratorEsGoodsIndexEvent("生成店铺商品", GoodsTagsEnum.GENERATOR_STORE_GOODS_INDEX.name(), storeId));
applicationEventPublisher.publishEvent(new TransactionCommitSendMQEvent("生成店铺商品", rocketmqCustomProperties.getGoodsTopic(), GoodsTagsEnum.GENERATOR_STORE_GOODS_INDEX.name(), storeId));
} else if (GoodsStatusEnum.DOWN.name().equals(marketEnable)) {
cache.vagueDel(CachePrefix.GOODS_SKU.getPrefix());
applicationEventPublisher.publishEvent(new GeneratorEsGoodsIndexEvent("删除店铺商品", GoodsTagsEnum.STORE_GOODS_DELETE.name(), storeId));
applicationEventPublisher.publishEvent(new TransactionCommitSendMQEvent("删除店铺商品", rocketmqCustomProperties.getGoodsTopic(), GoodsTagsEnum.STORE_GOODS_DELETE.name(), storeId));
}
}
}
@ -589,7 +589,7 @@ public class GoodsSkuServiceImpl extends ServiceImpl<GoodsSkuMapper, GoodsSku> i
if (!GoodsStatusEnum.UPPER.name().equals(goods.getMarketEnable()) || !GoodsAuthEnum.PASS.name().equals(goods.getAuthFlag())) {
return;
}
applicationEventPublisher.publishEvent(new GeneratorEsGoodsIndexEvent("生成商品", GoodsTagsEnum.GENERATOR_GOODS_INDEX.name(), goods.getId()));
applicationEventPublisher.publishEvent(new TransactionCommitSendMQEvent("生成商品", rocketmqCustomProperties.getGoodsTopic(), GoodsTagsEnum.GENERATOR_GOODS_INDEX.name(), goods.getId()));
}
@Override

31
framework/src/main/java/cn/lili/modules/order/order/serviceimpl/OrderServiceImpl.java
View File

@ -10,6 +10,7 @@ import cn.hutool.poi.excel.ExcelUtil;
import cn.hutool.poi.excel.ExcelWriter;
import cn.lili.common.enums.PromotionTypeEnum;
import cn.lili.common.enums.ResultCode;
import cn.lili.common.event.TransactionCommitSendMQEvent;
import cn.lili.common.exception.ServiceException;
import cn.lili.common.properties.RocketmqCustomProperties;
import cn.lili.common.security.OperationalJudgment;
@ -66,6 +67,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.util.CellRangeAddressList;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
@ -144,6 +146,10 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
@Autowired
private TradeService tradeService;
@Autowired
private ApplicationEventPublisher applicationEventPublisher;
@Override
@Transactional(rollbackFor = Exception.class)
public void intoDB(TradeDTO tradeDTO) {
@ -445,12 +451,13 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
//获取订单信息
Order order = this.getBySn(orderSn);
//获取踪迹信息
String str=order.getConsigneeMobile();
return logisticsService.getLogistic(order.getLogisticsCode(), order.getLogisticsNo(), str.substring(str.length()-4));
String str = order.getConsigneeMobile();
return logisticsService.getLogistic(order.getLogisticsCode(), order.getLogisticsNo(), str.substring(str.length() - 4));
}
@Override
@OrderLogPoint(description = "'订单['+#orderSn+']核销，核销码['+#verificationCode+']'", orderSn = "#orderSn")
@Transactional(rollbackFor = Exception.class)
public Order take(String orderSn, String verificationCode) {
//获取订单信息
@ -495,7 +502,8 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
* @param order 订单
* @param orderSn 订单编号
*/
private void complete(Order order, String orderSn) {//修改订单状态为完成
@Transactional(rollbackFor = Exception.class)
public void complete(Order order, String orderSn) {//修改订单状态为完成
this.updateStatus(orderSn, OrderStatusEnum.COMPLETED);
//修改订单货物可以进行评价
@ -534,10 +542,9 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
}
@Override
@Transactional(rollbackFor = Exception.class)
public void sendUpdateStatusMessage(OrderMessage orderMessage) {
String destination = rocketmqCustomProperties.getOrderTopic() + ":" + OrderTagsEnum.STATUS_CHANGE.name();
//发送订单变更mq消息
rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(orderMessage), RocketmqSendCallbackBuilder.commonCallback());
applicationEventPublisher.publishEvent(new TransactionCommitSendMQEvent("发送订单变更mq消息", rocketmqCustomProperties.getOrderTopic(), OrderTagsEnum.STATUS_CHANGE.name(), JSONUtil.toJsonStr(orderMessage)));
}
@Override
@ -621,6 +628,7 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
}
@Override
@Transactional(rollbackFor = Exception.class)
public void batchDeliver(MultipartFile files) {
InputStream inputStream;
@ -777,9 +785,10 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
/**
* 订单状态变更消息
*
* @param order
* @param order 订单信息
*/
private void orderStatusMessage(Order order) {
@Transactional(rollbackFor = Exception.class)
public void orderStatusMessage(Order order) {
OrderMessage orderMessage = new OrderMessage();
orderMessage.setOrderSn(order.getSn());
orderMessage.setNewStatus(OrderStatusEnum.valueOf(order.getOrderStatus()));
@ -906,7 +915,8 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
*
* @param orderSn 订单编号
*/
private void normalOrderConfirm(String orderSn) {
@Transactional(rollbackFor = Exception.class)
public void normalOrderConfirm(String orderSn) {
//修改订单
this.update(new LambdaUpdateWrapper<Order>()
.eq(Order::getSn, orderSn)
@ -925,7 +935,8 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
*
* @param orderSn 订单编号
*/
private void virtualOrderConfirm(String orderSn) {
@Transactional(rollbackFor = Exception.class)
public void virtualOrderConfirm(String orderSn) {
//修改订单
this.update(new LambdaUpdateWrapper<Order>()
.eq(Order::getSn, orderSn)

19
framework/src/main/java/cn/lili/modules/promotion/event/UpdateEsGoodsIndexPromotionsEvent.java
View File

@ -1,19 +0,0 @@
package cn.lili.modules.promotion.event;
import lombok.Data;
import org.springframework.context.ApplicationEvent;
/**
* @author paulG
* @since 2022/1/19
**/
@Data
public class UpdateEsGoodsIndexPromotionsEvent extends ApplicationEvent {
private String promotionsJsonStr;
public UpdateEsGoodsIndexPromotionsEvent(Object source, String promotionsJsonStr) {
super(source);
this.promotionsJsonStr = promotionsJsonStr;
}
}

41
framework/src/main/java/cn/lili/modules/promotion/listener/UpdateEsGoodsIndexPromotionsListener.java
View File

@ -1,41 +0,0 @@
package cn.lili.modules.promotion.listener;
import cn.lili.common.properties.RocketmqCustomProperties;
import cn.lili.modules.promotion.event.UpdateEsGoodsIndexPromotionsEvent;
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
import cn.lili.rocketmq.tags.GoodsTagsEnum;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
/**
* @author paulG
* @since 2022/1/19
**/
@Component
public class UpdateEsGoodsIndexPromotionsListener {
/**
* rocketMq
*/
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* rocketMq配置
*/
@Autowired
private RocketmqCustomProperties rocketmqCustomProperties;
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
public void updateEsGoodsIndexPromotions(UpdateEsGoodsIndexPromotionsEvent event) {
//更新商品促销消息
String destination = rocketmqCustomProperties.getGoodsTopic() + ":" + GoodsTagsEnum.UPDATE_GOODS_INDEX_PROMOTIONS.name();
//发送mq消息
rocketMQTemplate.asyncSend(destination, event.getPromotionsJsonStr(), RocketmqSendCallbackBuilder.commonCallback());
}
}

17
framework/src/main/java/cn/lili/modules/promotion/serviceimpl/AbstractPromotionsServiceImpl.java
View File

@ -4,6 +4,7 @@ import cn.hutool.core.map.MapBuilder;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.json.JSONUtil;
import cn.lili.common.enums.ResultCode;
import cn.lili.common.event.TransactionCommitSendMQEvent;
import cn.lili.common.exception.ServiceException;
import cn.lili.common.properties.RocketmqCustomProperties;
import cn.lili.common.vo.PageVO;
@ -11,19 +12,16 @@ import cn.lili.modules.promotion.entity.dos.BasePromotions;
import cn.lili.modules.promotion.entity.dos.PromotionGoods;
import cn.lili.modules.promotion.entity.dto.search.BasePromotionsSearchParams;
import cn.lili.modules.promotion.entity.enums.PromotionsScopeTypeEnum;
import cn.lili.modules.promotion.event.UpdateEsGoodsIndexPromotionsEvent;
import cn.lili.modules.promotion.service.AbstractPromotionsService;
import cn.lili.modules.promotion.service.PromotionGoodsService;
import cn.lili.modules.promotion.tools.PromotionTools;
import cn.lili.mybatis.util.PageUtil;
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
import cn.lili.rocketmq.tags.GoodsTagsEnum;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.transaction.annotation.Transactional;
@ -48,12 +46,6 @@ public abstract class AbstractPromotionsServiceImpl<M extends BaseMapper<T>, T e
@Autowired
private RocketmqCustomProperties rocketmqCustomProperties;
/**
* rocketMq
*/
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private ApplicationEventPublisher applicationEventPublisher;
@ -253,13 +245,12 @@ public abstract class AbstractPromotionsServiceImpl<M extends BaseMapper<T>, T e
* @param promotions 促销实体
*/
@Override
@Transactional(rollbackFor = {Exception.class})
public void updateEsGoodsIndex(T promotions) {
if (promotions.getStartTime() == null && promotions.getEndTime() == null) {
Map<Object, Object> build = MapBuilder.create().put("promotionKey", this.getPromotionType() + "-" + promotions.getId()).put("scopeId", promotions.getScopeId()).build();
//删除商品促销消息
String destination = rocketmqCustomProperties.getGoodsTopic() + ":" + GoodsTagsEnum.DELETE_GOODS_INDEX_PROMOTIONS.name();
//发送mq消息
rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(build), RocketmqSendCallbackBuilder.commonCallback());
applicationEventPublisher.publishEvent(new TransactionCommitSendMQEvent("删除商品促销事件", rocketmqCustomProperties.getGoodsTopic(), GoodsTagsEnum.DELETE_GOODS_INDEX_PROMOTIONS.name(), JSONUtil.toJsonStr(build)));
} else {
String esPromotionKey = this.getPromotionType().name() + "-" + promotions.getId();
@ -270,7 +261,7 @@ public abstract class AbstractPromotionsServiceImpl<M extends BaseMapper<T>, T e
map.put("promotionsType", promotions.getClass().getName());
// 促销实体
map.put("promotions", promotions);
applicationEventPublisher.publishEvent(new UpdateEsGoodsIndexPromotionsEvent("更新商品索引促销事件", JSONUtil.toJsonStr(map)));
applicationEventPublisher.publishEvent(new TransactionCommitSendMQEvent("更新商品索引促销事件", rocketmqCustomProperties.getGoodsTopic(), GoodsTagsEnum.UPDATE_GOODS_INDEX_PROMOTIONS.name(), JSONUtil.toJsonStr(map)));
}
}

4
framework/src/main/java/cn/lili/modules/wallet/serviceimpl/MemberWalletServiceImpl.java
View File

@ -84,7 +84,7 @@ public class MemberWalletServiceImpl extends ServiceImpl<MemberWalletMapper, Mem
QueryWrapper<MemberWallet> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("member_id", memberId);
//执行查询
MemberWallet memberWallet = this.baseMapper.selectOne(queryWrapper);
MemberWallet memberWallet = this.getOne(queryWrapper, false);
//如果没有钱包，则创建钱包
if (memberWallet == null) {
memberWallet = this.save(memberId, memberService.getById(memberId).getUsername());
@ -186,7 +186,7 @@ public class MemberWalletServiceImpl extends ServiceImpl<MemberWalletMapper, Mem
*/
private MemberWallet checkMemberWallet(String memberId) {
//获取会员预存款信息
MemberWallet memberWallet = this.getOne(new QueryWrapper<MemberWallet>().eq("member_id", memberId));
MemberWallet memberWallet = this.getOne(new QueryWrapper<MemberWallet>().eq("member_id", memberId), false);
//如果会员预存款信息不存在则同步重新建立预存款信息
if (memberWallet == null) {
Member member = memberService.getById(memberId);