调整异步方法;添加异步库存同步

This commit is contained in:
DB 2023-11-01 20:56:36 +08:00
parent 6652f37473
commit 0cd334cac0
6 changed files with 50 additions and 47 deletions

View File

@ -3,6 +3,7 @@ package com.cpop.core.config;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
@ -14,6 +15,7 @@ import java.util.concurrent.ThreadPoolExecutor;
* @Description: * @Description:
*/ */
@Configuration @Configuration
@EnableAsync
public class AsyncScheduledTaskConfig { public class AsyncScheduledTaskConfig {
@Autowired @Autowired

View File

@ -3,7 +3,6 @@ package com.cpop.mall.web;
import org.mybatis.spring.annotation.MapperScan; import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication; import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.EnableScheduling;
/** /**
@ -11,7 +10,6 @@ import org.springframework.scheduling.annotation.EnableScheduling;
*/ */
@SpringBootApplication(scanBasePackages = {"com.cpop.**"}) @SpringBootApplication(scanBasePackages = {"com.cpop.**"})
@MapperScan("com.cpop.**.mapper") @MapperScan("com.cpop.**.mapper")
@EnableAsync
@EnableScheduling @EnableScheduling
public class CpopMallWebApplication { public class CpopMallWebApplication {

View File

@ -31,7 +31,7 @@ spring:
max-file-size: 1024MB max-file-size: 1024MB
max-request-size: 300MB max-request-size: 300MB
profiles: profiles:
active: dev,core,mall,system active: local,core,mall,system
datasource: datasource:
type: com.zaxxer.hikari.HikariDataSource type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver driver-class-name: com.mysql.cj.jdbc.Driver

View File

@ -26,7 +26,6 @@ import com.mybatisflex.core.paginate.Page;
import com.mybatisflex.core.query.QueryWrapper; import com.mybatisflex.core.query.QueryWrapper;
import com.mybatisflex.spring.service.impl.ServiceImpl; import com.mybatisflex.spring.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import static com.cpop.core.base.table.table.SysUserTableDef.SYS_USER; import static com.cpop.core.base.table.table.SysUserTableDef.SYS_USER;
@ -162,6 +161,7 @@ public class OrderRefundServiceImpl extends ServiceImpl<OrderRefundMapper, Order
this.updateChain().set(ORDER_REFUND.REFUND_STATUS, 1) this.updateChain().set(ORDER_REFUND.REFUND_STATUS, 1)
.set(ORDER_REFUND.OUT_REFUND_ID, reqInfo.getRefundId()) .set(ORDER_REFUND.OUT_REFUND_ID, reqInfo.getRefundId())
.where(ORDER_REFUND.ID.eq(reqInfo.getOutRefundNo())).update(); .where(ORDER_REFUND.ID.eq(reqInfo.getOutRefundNo())).update();
//TODO:回滚库存
} catch (WxPayException e) { } catch (WxPayException e) {
throw new ServiceException(e.getMessage()); throw new ServiceException(e.getMessage());
} }

View File

@ -20,6 +20,7 @@ import com.cpop.mall.business.dto.WxPayGoodsDetailDto;
import com.cpop.mall.business.entity.*; import com.cpop.mall.business.entity.*;
import com.cpop.mall.business.mapper.OrderMapper; import com.cpop.mall.business.mapper.OrderMapper;
import com.cpop.mall.business.service.*; import com.cpop.mall.business.service.*;
import com.cpop.mall.business.task.ProductRecordSyncStockTask;
import com.cpop.mall.business.vo.OrderPageVo; import com.cpop.mall.business.vo.OrderPageVo;
import com.cpop.mall.framework.config.wxPay.WxPayProperties; import com.cpop.mall.framework.config.wxPay.WxPayProperties;
import com.cpop.mall.framework.constant.MallRedisConstant; import com.cpop.mall.framework.constant.MallRedisConstant;
@ -35,10 +36,8 @@ import com.github.binarywang.wxpay.exception.WxPayException;
import com.github.binarywang.wxpay.service.WxPayService; import com.github.binarywang.wxpay.service.WxPayService;
import com.mybatisflex.core.paginate.Page; import com.mybatisflex.core.paginate.Page;
import com.mybatisflex.core.query.QueryWrapper; import com.mybatisflex.core.query.QueryWrapper;
import com.mybatisflex.core.row.Db;
import com.mybatisflex.spring.service.impl.ServiceImpl; import com.mybatisflex.spring.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
@ -323,8 +322,6 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
productRecords.forEach(item -> { productRecords.forEach(item -> {
item.setRecordNum(item.getRecordNum() - orderNumMap.get(item.getId())); item.setRecordNum(item.getRecordNum() - orderNumMap.get(item.getId()));
}); });
//异步更新
asyncUpdateRecords(productRecords, orderNumMap);
//TODO:分账先注释 //TODO:分账先注释
ProfitSharing profitSharing = new ProfitSharing(); ProfitSharing profitSharing = new ProfitSharing();
profitSharing.setOrderId(orderId) profitSharing.setOrderId(orderId)
@ -353,50 +350,13 @@ public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements
profitSharingService.updateChain().set(PROFIT_SHARING.OUT_PROFIT_SHARING_ID,profitSharingResult.getOrderId()) profitSharingService.updateChain().set(PROFIT_SHARING.OUT_PROFIT_SHARING_ID,profitSharingResult.getOrderId())
.set(PROFIT_SHARING.AMOUNT, ceil.longValue()).set(PROFIT_SHARING.PROFIT_SHARING_STATUS, 1).set(PROFIT_SHARING.PAY_ACCOUNT, wxPayProperties.getSharingAccount()) .set(PROFIT_SHARING.AMOUNT, ceil.longValue()).set(PROFIT_SHARING.PROFIT_SHARING_STATUS, 1).set(PROFIT_SHARING.PAY_ACCOUNT, wxPayProperties.getSharingAccount())
.where(PROFIT_SHARING.ID.eq(profitSharing.getId())).update(); .where(PROFIT_SHARING.ID.eq(profitSharing.getId())).update();
//异步更新
SpringUtils.getBean(ProductRecordSyncStockTask.class).asyncUpdateRecords(productRecords, orderNumMap);
} catch (WxPayException e) { } catch (WxPayException e) {
throw new ServiceException(e.getMessage()); throw new ServiceException(e.getMessage());
} }
} }
/**
* @Description: 异步更新数据库
* @param productRecords 产品记录
* @Author DB
* @Date: 2023/11/1 0:09
*/
@Async("customAsyncThreadPool")
public void asyncUpdateRecords(List<ProductRecord> productRecords,Map<String, Integer> orderNumMap) {
loopUpdateStock(productRecords, orderNumMap);
}
/**
* @descriptions 循环插入库存修改记录
* @author DB
* @date 2023/11/01 17:01
* @param productRecords 库存修改记录
* @return: void
*/
private void loopUpdateStock(List<ProductRecord> productRecords,Map<String, Integer> orderNumMap) {
ProductRecordService productRecordService = SpringUtils.getBean(ProductRecordService.class);
//用迭代器
Iterator<ProductRecord> iterator = productRecords.iterator();
while (iterator.hasNext()) {
ProductRecord next = iterator.next();
next.setRecordNum(next.getRecordNum() - orderNumMap.get(next.getId()));
boolean update = productRecordService.updateById(next);
//如果更新成功移除
if (update) {
iterator.remove();
}
}
//存在更新失败,需要重新更新,直到全部成功
if (!productRecords.isEmpty()){
//获取最新数据进行下一次循环
loopUpdateStock(productRecordService.listByIds(productRecords.stream().map(ProductRecord::getId).collect(Collectors.toSet())),
orderNumMap);
}
}
/** /**
* @descriptions 取消订单 * @descriptions 取消订单
* @author DB * @author DB

View File

@ -6,10 +6,14 @@ import com.cpop.mall.business.entity.ProductRecord;
import com.cpop.mall.business.service.ProductRecordService; import com.cpop.mall.business.service.ProductRecordService;
import com.cpop.mall.framework.constant.MallRedisConstant; import com.cpop.mall.framework.constant.MallRedisConstant;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/** /**
* @author DB * @author DB
@ -38,4 +42,43 @@ public class ProductRecordSyncStockTask {
}); });
log.info("===============================结束<--库存信息结束-->结束==============================="); log.info("===============================结束<--库存信息结束-->结束===============================");
} }
/**
* @Description: 异步更新数据库
* @param productRecords 产品记录
* @Author DB
* @Date: 2023/11/1 0:09
*/
@Async("customAsyncThreadPool")
public void asyncUpdateRecords(List<ProductRecord> productRecords, Map<String, Integer> orderNumMap) {
loopUpdateStock(productRecords, orderNumMap);
}
/**
* @descriptions 循环插入库存修改记录
* @author DB
* @date 2023/11/01 17:01
* @param productRecords 库存修改记录
* @return: void
*/
private void loopUpdateStock(List<ProductRecord> productRecords,Map<String, Integer> orderNumMap) {
ProductRecordService productRecordService = SpringUtils.getBean(ProductRecordService.class);
//用迭代器
Iterator<ProductRecord> iterator = productRecords.iterator();
while (iterator.hasNext()) {
ProductRecord next = iterator.next();
next.setRecordNum(next.getRecordNum() - orderNumMap.get(next.getId()));
boolean update = productRecordService.updateById(next);
//如果更新成功移除
if (update) {
iterator.remove();
}
}
//存在更新失败(乐观锁),需要重新更新,直到全部成功
if (!productRecords.isEmpty()){
//获取最新数据进行下一次循环
loopUpdateStock(productRecordService.listByIds(productRecords.stream().map(ProductRecord::getId).collect(Collectors.toSet())),
orderNumMap);
}
}
} }