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-14 12:25:53 +08:00
parent 25dc59e213
commit 1100878b92
6 changed files with 318 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

242
mall-common/src/main/java/com/suisung/mall/common/service/impl/MessageServiceImpl.java
View File

@ -5,6 +5,7 @@ import com.suisung.mall.common.modules.admin.AdminLogError;
import com.suisung.mall.common.modules.analytics.AnalyticsAccessHistory; import com.suisung.mall.common.modules.analytics.AnalyticsAccessHistory;
import com.suisung.mall.common.service.MessageService; import com.suisung.mall.common.service.MessageService;
import com.suisung.mall.core.web.service.RedisService; import com.suisung.mall.core.web.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@ -12,6 +13,7 @@ import java.util.Date;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
@Slf4j
@Service @Service
public class MessageServiceImpl implements MessageService { public class MessageServiceImpl implements MessageService {
private static final String kerPre = "_msg_|"; private static final String kerPre = "_msg_|";
@ -35,99 +37,237 @@ public class MessageServiceImpl implements MessageService {
* @param user_id 用户ID * @param user_id 用户ID
* @param store_id 消息模板ID * @param store_id 消息模板ID
* @param message_tpl_id 消息模板ID * @param message_tpl_id 消息模板ID
* @return args 参数 * @param args 参数
* @return 是否发送成功
* @access public * @access public
*/ */
@Override @Override
public boolean sendNoticeMsg(Integer user_id, Integer store_id, String message_tpl_id, Map args) { public boolean sendNoticeMsg(Integer user_id, Integer store_id, String message_tpl_id, Map args) {
//进入异步队列 try {
String key = getKey("notice_msg"); // 参数校验
if (message_tpl_id == null || message_tpl_id.trim().isEmpty()) {
log.warn("消息模板ID为空，无法发送通知消息");
return false;
}
//进入异步队列
String key = getKey("notice_msg");
Map<String, Object> msg_data = new HashMap<>(); Map<String, Object> msg_data = new HashMap<>();
msg_data.put("user_id", user_id); msg_data.put("user_id", user_id);
msg_data.put("store_id", store_id); msg_data.put("store_id", store_id);
msg_data.put("message_tpl_id", message_tpl_id); msg_data.put("message_tpl_id", message_tpl_id);
msg_data.put("args", args); msg_data.put("args", args);
msg_data.put("time", new Date().getTime()); msg_data.put("time", new Date().getTime());
Long res = redisService.lPush(key, JSONUtil.toJsonStr(msg_data)); // 检查Redis服务是否可用
if (redisService == null) {
log.error("Redis服务未初始化");
return false;
}
String jsonData = JSONUtil.toJsonStr(msg_data);
if (jsonData == null) {
log.warn("通知消息对象序列化为JSON失败");
return false;
}
Long res = redisService.lPush(key, jsonData);
// 检查返回值是否为空
if (res == null) {
log.warn("Redis lPush操作返回空值");
return false;
}
return res > 0; return res > 0;
} catch (Exception e) {
log.error("发送通知消息到队列时发生异常, user_id: {}, store_id: {}, message_tpl_id: {}",
user_id, store_id, message_tpl_id, e);
return false;
}
} }
//示例，需要移动到计划任务中。 /**
* 从队列中读取通知消息
*
* @return 通知消息数据的JSON字符串，如果没有数据或发生异常则返回null
*/
@Override @Override
public String receiveNoticeMsg() { public String receiveNoticeMsg() {
//进入异步队列 try {
String key = getKey("notice_msg"); //进入异步队列
String data = null; String key = getKey("notice_msg");
// 检查Redis服务是否可用
if (redisService == null) {
log.error("Redis服务未初始化");
return null;
}
Integer times = 100000; //如果队列有数据，一直读完
Long queueSize = redisService.lSize(key);
//如果队列有数据，一直读完 if (queueSize != null && queueSize > 0) {
if (redisService.lSize(key) > 0) { Object result = redisService.rPop(key);
data = redisService.rPop(key).toString(); if (result != null) {
return result.toString();
}
}
} catch (Exception e) {
log.error("从队列读取通知消息时发生异常", e);
} }
return data; return null;
} }
/**
* 发送访问日志到队列
*
* @param accessDTO 访问日志对象
* @return 是否发送成功
*/
@Override @Override
public boolean sendAccess(AnalyticsAccessHistory accessDTO) { public boolean sendAccess(AnalyticsAccessHistory accessDTO) {
//进入异步队列 try {
String key = getKey("access"); //进入异步队列
if (accessDTO == null) {
Long res = redisService.lPush(key, JSONUtil.toJsonStr(accessDTO)); log.warn("尝试发送空的访问日志对象");
return false;
return res > 0; }
String key = getKey("access");
// 检查Redis服务是否可用
if (redisService == null) {
log.error("Redis服务未初始化");
return false;
}
String jsonData = JSONUtil.toJsonStr(accessDTO);
if (jsonData == null) {
log.warn("访问日志对象序列化为JSON失败");
return false;
}
Long res = redisService.lPush(key, jsonData);
// 检查返回值是否为空
if (res == null) {
log.warn("Redis lPush操作返回空值");
return false;
}
return res > 0;
} catch (Exception e) {
log.error("发送访问日志到队列时发生异常", e);
return false;
}
} }
/**
* 从队列中读取访问日志数据
*
* @return 访问日志数据的JSON字符串，如果没有数据或发生异常则返回null
*/
@Override @Override
public String receiveAccess() { public String receiveAccess() {
//进入异步队列 try {
String key = getKey("access"); //进入异步队列
String data = null; String key = getKey("access");
// 检查Redis服务是否可用
if (redisService == null) {
log.error("Redis服务未初始化");
return null;
}
//如果队列有数据，一直读完 //如果队列有数据，一直读完
if (redisService.lSize(key) > 0) { Long queueSize = redisService.lSize(key);
data = redisService.rPop(key).toString(); if (queueSize != null && queueSize > 0) {
Object result = redisService.rPop(key);
if (result != null) {
return result.toString();
}
}
} catch (Exception e) {
log.error("从队列读取访问日志数据时发生异常", e);
} }
return data; return null;
} }
/** /**
* 消息入队列 * 消息入队列
* *
* @param logError * @param logError 错误日志对象
* @return * @return 是否发送成功
*/ */
@Override @Override
public boolean sendError(AdminLogError logError) { public boolean sendError(AdminLogError logError) {
//进入异步队列 try {
String key = getKey("error"); //进入异步队列
if (logError == null) {
log.warn("尝试发送空的错误日志对象");
return false;
}
String key = getKey("error");
// 检查Redis服务是否可用
if (redisService == null) {
log.error("Redis服务未初始化");
return false;
}
String jsonData = JSONUtil.toJsonStr(logError);
if (jsonData == null) {
log.warn("错误日志对象序列化为JSON失败");
return false;
}
Long res = redisService.lPush(key, jsonData);
// 检查返回值是否为空
if (res == null) {
log.warn("Redis lPush操作返回空值");
return false;
}
Long res = redisService.lPush(key, JSONUtil.toJsonStr(logError)); return res > 0;
} catch (Exception e) {
return res > 0; log.error("发送错误日志到队列时发生异常", e);
return false;
}
} }
/** /**
* 从队列中读取 * 从队列中读取错误日志数据
* *
* @return * @return 错误日志数据的JSON字符串，如果没有数据或发生异常则返回null
*/ */
@Override @Override
public String receiveError() { public String receiveError() {
//进入异步队列 try {
String key = getKey("error"); String key = getKey("error");
String data = null;
// 检查Redis服务是否可用
if (redisService == null) {
log.error("Redis服务未初始化");
return null;
}
//如果队列有数据，一直读完 // 检查队列是否存在且不为空
if (redisService.lSize(key) > 0) { Long queueSize = redisService.lSize(key);
data = redisService.rPop(key).toString(); if (queueSize != null && queueSize > 0) {
Object result = redisService.rPop(key);
if (result != null) {
return result.toString();
}
}
} catch (Exception e) {
log.error("从队列读取错误日志数据时发生异常", e);
} }
return data; return null;
} }
} }

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

@ -18,6 +18,7 @@ import com.suisung.mall.shop.message.service.MqMessageService;
import com.suisung.mall.shop.message.service.PushMessageService; import com.suisung.mall.shop.message.service.PushMessageService;
import com.suisung.mall.shop.order.service.ShopOrderBaseService; import com.suisung.mall.shop.order.service.ShopOrderBaseService;
import com.suisung.mall.shop.order.service.ShopOrderReturnService; import com.suisung.mall.shop.order.service.ShopOrderReturnService;
import com.suisung.mall.shop.sfexpress.service.SFExpressApiService;
import com.suisung.mall.shop.store.service.ShopStoreSameCityTransportBaseService; import com.suisung.mall.shop.store.service.ShopStoreSameCityTransportBaseService;
import io.swagger.annotations.Api; import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
@ -62,6 +63,10 @@ public class LakalaController extends BaseControllerImpl {
@Resource @Resource
private ShopOrderBaseService shopOrderBaseService; private ShopOrderBaseService shopOrderBaseService;
@Lazy
@Resource
private SFExpressApiService sfExpressApiService;
@Lazy @Lazy
@Resource @Resource
private MqMessageService mqMessageService; private MqMessageService mqMessageService;
@ -98,7 +103,9 @@ public class LakalaController extends BaseControllerImpl {
// mqMessageService.sendDelayMessage(jsonObject.toString(), 10000); // mqMessageService.sendDelayMessage(jsonObject.toString(), 10000);
// return jsonObject; // return jsonObject;
return shopOrderBaseService.sameCityOrderExpireSeconds(10000L); // return shopOrderBaseService.sameCityOrderExpireSeconds(10000L);
return sfExpressApiService.createSfExpressShop(58, "桂平能辉超市", "桂平市", "广西壮族自治区贵港市桂平市中山南路凤凰商业中心19-3号", "谢能坤", "17777525395", "110.08105", "23.39339");
} }

2
mall-shop/src/main/java/com/suisung/mall/shop/sfexpress/service/SFExpressApiService.java
View File

@ -39,7 +39,7 @@ public interface SFExpressApiService {
* @param sfShopId 商家订单号 * @param sfShopId 商家订单号
* @return * @return
*/ */
ThirdApiRes getShopInfo(String sfShopId); ThirdApiRes getSfShopInfo(String sfShopId);
/** /**
* 内部顺丰同城订单下单 * 内部顺丰同城订单下单

138
mall-shop/src/main/java/com/suisung/mall/shop/sfexpress/service/impl/SFExpressApiServiceImpl.java
View File

@ -94,7 +94,7 @@ public class SFExpressApiServiceImpl implements SFExpressApiService {
/** /**
* 创建顺丰同店铺-连锁店铺 * 创建顺丰同城（普通型）店铺
* *
* @param storeId 商家门店ID * @param storeId 商家门店ID
* @param shopName 店名 * @param shopName 店名
@ -104,88 +104,93 @@ public class SFExpressApiServiceImpl implements SFExpressApiService {
* @param contactPhone 店铺电话 * @param contactPhone 店铺电话
* @param longitude 经度 * @param longitude 经度
* @param latitude 纬度 * @param latitude 纬度
* @return * @return Pair<Boolean, String> 第一个元素表示是否成功，第二个元素表示结果信息或错误信息
*/ */
@Override @Override
public Pair<Boolean, String> createSfExpressShop(Integer storeId, String shopName, String cityName, String shopAddress, String contactName, String contactPhone, String longitude, String latitude) { public Pair<Boolean, String> createSfExpressShop(Integer storeId, String shopName, String cityName,
logger.info("开始创建顺丰同城店铺"); String shopAddress, String contactName, String contactPhone, String longitude, String latitude) {
logger.info("开始创建顺丰同城店铺, storeId: {}", storeId);
if (CheckUtil.isEmpty(storeId) && StringUtils.isAnyBlank(shopName, shopAddress, contactName, contactPhone)) { // 验证必要参数
if (CheckUtil.isEmpty(storeId) || StringUtils.isAnyBlank(shopName, shopAddress, contactName, contactPhone)) {
return Pair.of(false, "顺丰同城店铺，缺少必要参数！"); return Pair.of(false, "顺丰同城店铺，缺少必要参数！");
} }
ShopStoreSameCityTransportBase shopStoreSameCityTransportBase = shopStoreSameCityTransportBaseService.getShopStoreSameCityTransportBaseById(Long.valueOf(storeId)); // 获取或初始化商家配送信息
if (shopStoreSameCityTransportBase == null) { ShopStoreSameCityTransportBase transportBase = shopStoreSameCityTransportBaseService
// 如果没有商家配送运费设置（包含有顺丰店铺ID字段在里面），则初始化 .getShopStoreSameCityTransportBaseById(Long.valueOf(storeId));
Pair<Boolean, String> result = shopStoreSameCityTransportBaseService.initDefaultSameCityTransport(storeId);
if (!result.getFirst()) { if (transportBase == null) {
return result; // 如果没有商家配送运费设置，则初始化
Pair<Boolean, String> initResult = shopStoreSameCityTransportBaseService.initDefaultSameCityTransport(storeId);
if (!initResult.getFirst()) {
return initResult;
} }
transportBase = shopStoreSameCityTransportBaseService.getShopStoreSameCityTransportBaseById(Long.valueOf(storeId));
shopStoreSameCityTransportBase = shopStoreSameCityTransportBaseService.getShopStoreSameCityTransportBaseById(Long.valueOf(storeId));
} }
if (CheckUtil.isNotEmpty(shopStoreSameCityTransportBase.getShop_id())) { // 如果已存在顺丰店铺ID，验证其有效性
return Pair.of(true, "顺丰同城已创建过该店铺！"); // if (CheckUtil.isNotEmpty(transportBase.getShop_id())) {
} // ThirdApiRes shopInfo = getSfShopInfo(transportBase.getShop_id());
// if (shopInfo != null && shopInfo.getError_code().equals(0)) {
// return Pair.of(true, transportBase.getShop_id());
// }
// }
// 构建请求参数：
Map<String, Object> params = buildCommonParams(); Map<String, Object> params = buildCommonParams();
params.put("supplier_id", supplierId); params.put("supplier_id", supplierId); // 店铺所属商家id
params.put("out_shop_id", storeId); params.put("out_shop_id", storeId); // 外部店铺ID
params.put("shop_name", shopName); //店铺名称不可重复 params.put("shop_name", shopName); // 店铺名称
params.put("city_name", cityName); params.put("city_name", cityName); // 城市名称
//1:快餐 2:药品 3:百货 4:脏衣服收 5:干净衣服派 6:生鲜 8:高端饮品 9:现场勘验 10:快递 12:文件 13:蛋糕 14:鲜花 15:数码 16:服装 17: //1:快餐 2:药品 3:百货 4:脏衣服收 5:干净衣服派 6:生鲜 8:高端饮品 9:现场勘验 10:快递 12:文件 13:蛋糕 14:鲜花 15:数码 16:服装 17:
//汽配 18:珠宝 20:披萨 21:中餐 22:水产 27:专人直送 32:中端饮品 33:便利店 34:面包糕点 35:火锅 36:证照 40:烧烤小龙虾 41:外部落地配 47:烟酒 //汽配 18:珠宝 20:披萨 21:中餐 22:水产 27:专人直送 32:中端饮品 33:便利店 34:面包糕点 35:火锅 36:证照 40:烧烤小龙虾 41:外部落地配 47:烟酒
//行 48:成人用品 99:其他 //行 48:成人用品 99:其他
params.put("shop_product_types", "1,3"); // 店铺经营类型（支持多品类，枚举值见下方） params.put("shop_product_types", "1,3,6,8,13,14,15,16,20,22,32,33,34,47,99"); // 经营类型: 快餐,百货,生鲜,便利店
params.put("shop_type", 1); // 店铺类型: 1-普通型 2-平台型
params.put("shop_address", shopAddress); // 店铺地址
params.put("longitude", longitude); // 经度
params.put("latitude", latitude); // 纬度
params.put("shop_contact_name", contactName); // 联系人姓名
params.put("shop_contact_phone", contactPhone); // 联系电话
params.put("shop_type", 1);// 店铺类型:1-普通型 2-平台型 // 发送请求到顺丰接口
params.put("shop_address", shopAddress);
params.put("longitude", longitude);
params.put("latitude", latitude);
params.put("shop_contact_name", contactName);
params.put("shop_contact_phone", contactPhone);
// 请求参数转换 json 字符串参数
String paramJSON = JsonUtil.toJSONString(params); String paramJSON = JsonUtil.toJSONString(params);
// 根据参数生成请求签名 String sendUrl = buildUrl("createShop", paramJSON);
String send_url = buildUrl("createShop", paramJSON); String responseStr = HttpUtil.post(sendUrl, paramJSON);
// 向顺丰同城 创建一个顺丰同城配送订单 if (StrUtil.isBlank(responseStr)) {
String retRespStr = HttpUtil.post(send_url, paramJSON);
if (StrUtil.isEmpty(retRespStr)) {
logger.error("创建顺丰同城店铺异常，无返回值！"); logger.error("创建顺丰同城店铺异常，无返回值！");
return Pair.of(false, "创建顺丰同城店铺异常，无返回值！"); return Pair.of(false, "创建顺丰同城店铺异常，无返回值！");
} }
JSONObject sfExpressApiRes = JSONUtil.parseObj(retRespStr); ThirdApiRes apiRes = JSONUtil.toBean(responseStr, ThirdApiRes.class);
if (sfExpressApiRes == null) { if (apiRes == null) {
logger.error("创建顺丰同城店铺异常，返回值有误！！"); logger.error("创建顺丰同城店铺异常，返回值有误！！");
return Pair.of(false, "创建顺丰同城店铺异常，返回值有误！"); return Pair.of(false, "创建顺丰同城店铺异常，返回值有误！");
} }
// 检查接口调用结果
if (!sfExpressApiRes.get("error_code").equals(0) || sfExpressApiRes.get("result") == null) { if (!apiRes.getError_code().equals(0) || apiRes.getResult() == null) {
logger.error("创建顺丰同城店铺失败: {}", sfExpressApiRes.get("error_msg")); String errMsg = apiRes.getError_code().equals(0) ? "创建顺丰同城店铺失败！" : apiRes.getError_msg();
return Pair.of(false, Convert.toStr(sfExpressApiRes.get("error_msg"))); logger.error("创建顺丰同城店铺失败: {}", errMsg);
return Pair.of(false, errMsg);
} }
// TODO 判断状态 // 提取顺丰店铺ID并更新数据库
JSONObject result = (JSONObject) apiRes.getResult();
String sfShopId = result.getStr("shop_id");
transportBase.setShop_id(sfShopId);
String sfShopId = sfExpressApiRes.getByPath("result.shop_id").toString(); Pair<Long, String> updateResult = shopStoreSameCityTransportBaseService
.saveOrUpdateShopStoreSameCityTransportBase(transportBase);
shopStoreSameCityTransportBase.setShop_id(sfShopId); if (updateResult.getFirst() <= 0) {
shopStoreSameCityTransportBaseService.saveOrUpdateShopStoreSameCityTransportBase(shopStoreSameCityTransportBase); logger.error("更新店铺信息失败");
return Pair.of(false, "更新店铺信息失败");
}
// 个推推送消息 return Pair.of(true, sfShopId);
// JSONObject payload = new JSONObject();
// payload.put("category", CommonConstant.PUSH_MSG_CATE_CREATE_SF_SHOP);
// payload.put("sfShopId", sfShopId);
// pushMessageService.sendMessage(null, sfShopId, "您有一笔新的订单", "您有一笔同城订单[" + shopOrderId + "]，请及时处理。", payload);
//
return Pair.of(true, "创建顺丰同城店铺成功！");
} }
@Override @Override
@ -269,7 +274,7 @@ public class SFExpressApiServiceImpl implements SFExpressApiService {
* @return * @return
*/ */
@Override @Override
public ThirdApiRes getShopInfo(String sfShopId) { public ThirdApiRes getSfShopInfo(String sfShopId) {
// 请求参数转换 json 字符串参数 // 请求参数转换 json 字符串参数
Map<String, Object> paramMap = buildCommonParams(); Map<String, Object> paramMap = buildCommonParams();
@ -277,14 +282,23 @@ public class SFExpressApiServiceImpl implements SFExpressApiService {
// 根据参数生成请求签名 // 根据参数生成请求签名
String send_url = buildUrl("getshopinfo", JSONUtil.toJsonStr(paramMap)); String send_url = buildUrl("getshopinfo", JSONUtil.toJsonStr(paramMap));
// // 向顺丰同城 创建一个顺丰同城配送订单 // 向顺丰同城 创建一个顺丰同城配送订单
// String retRespStr = HttpUtil.post(send_url, paramJSON); String retRespStr = HttpUtil.post(send_url, JSONUtil.toJsonStr(paramMap));
// if (StrUtil.isEmpty(retRespStr)) { if (StrUtil.isBlank(retRespStr)) {
// logger.error("创建顺丰同城订单异常，无返回值！"); logger.error("查询顺丰店铺信息，无返回值！");
// return Pair.of(false, "顺丰同城下单异常，无返回值！"); return null;
// } }
return null; ThirdApiRes thirdApiRes = JSONUtil.toBean(retRespStr, ThirdApiRes.class);
if (thirdApiRes == null && thirdApiRes.getResult() == null) {
logger.error("顺丰店铺信息有误！");
if (!thirdApiRes.getError_code().equals(0) && StrUtil.isNotEmpty(thirdApiRes.getError_msg())) {
logger.error(thirdApiRes.getError_msg());
}
return null;
}
return thirdApiRes;
} }
/** /**
@ -321,7 +335,7 @@ public class SFExpressApiServiceImpl implements SFExpressApiService {
// 向顺丰同城 创建一个顺丰同城配送订单 // 向顺丰同城 创建一个顺丰同城配送订单
String retRespStr = HttpUtil.post(send_url, paramJSON); String retRespStr = HttpUtil.post(send_url, paramJSON);
if (StrUtil.isEmpty(retRespStr)) { if (StrUtil.isBlank(retRespStr)) {
logger.error("创建顺丰同城订单异常，无返回值！"); logger.error("创建顺丰同城订单异常，无返回值！");
return Pair.of(false, "顺丰同城下单异常，无返回值！"); return Pair.of(false, "顺丰同城下单异常，无返回值！");
} }

77
mall-shop/src/main/java/com/suisung/mall/shop/sync/listen/ShopBatchSubmitListener.java
View File

@ -15,11 +15,14 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.util.*; import java.util.ArrayList;
import java.util.concurrent.*; import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@Component @Component
@ -27,21 +30,18 @@ import java.util.stream.Collectors;
public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelExcel> { public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelExcel> {
// 批处理阈值 // 批处理阈值
private static final int BATCH_SIZE = 500; private static final int BATCH_SIZE = 500;
// 数据缓存
private List<SxGoosModelExcel> cachedDataList = new ArrayList<>(BATCH_SIZE);
private SyncThirdDataService syncThirdDataService;
// 线程池配置 // 线程池配置
private final ExecutorService executorService; private final ExecutorService executorService;
// 数据缓存
private List<Future<?>> futures ; private final List<SxGoosModelExcel> cachedDataList = new ArrayList<>(BATCH_SIZE);
private AtomicInteger success; private final SyncThirdDataService syncThirdDataService;
private AtomicInteger fails; private final List<Future<?>> futures;
private AtomicInteger batchSize; private final AtomicInteger success;
private final AtomicInteger fails;
private final AtomicInteger batchSize;
@Setter @Setter
@Getter @Getter
private String storeId; private String storeId;
@Setter @Setter
@Getter @Getter
@ -49,19 +49,19 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
@Setter @Setter
@Getter @Getter
private Map<String,Integer> brandMaps; private Map<String, Integer> brandMaps;
public ShopBatchSubmitListener(SyncThirdDataService syncThirdDataService) { public ShopBatchSubmitListener(SyncThirdDataService syncThirdDataService) {
this.syncThirdDataService = syncThirdDataService; this.syncThirdDataService = syncThirdDataService;
// 创建线程池（根据CPU核心数优化） // 创建线程池（根据CPU核心数优化）
int corePoolSize = Runtime.getRuntime().availableProcessors(); int corePoolSize = Runtime.getRuntime().availableProcessors();
log.info("核心线程数量{}" , corePoolSize); // log.info("核心线程数量{}", corePoolSize);
this.executorService = Executors.newFixedThreadPool(6); this.executorService = Executors.newFixedThreadPool(corePoolSize);
this.futures = new ArrayList<>(); this.futures = new ArrayList<>();
this.success = new AtomicInteger(); this.success = new AtomicInteger();
this.fails = new AtomicInteger(); this.fails = new AtomicInteger();
this.batchSize= new AtomicInteger(); this.batchSize = new AtomicInteger();
} }
@Override @Override
@ -91,13 +91,13 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
// 等待所有任务完成 // 等待所有任务完成
for (Future<?> future : futures) { for (Future<?> future : futures) {
try { try {
log.info("任务结果:{}" ,future.get()); log.info("任务结果:{}", future.get());
} catch (Exception e) { } catch (Exception e) {
log.info("任务执行异常: {}", e.getMessage()); log.info("任务执行异常: {}", e.getMessage());
} }
} }
log.info("Excel解析完成，总处理条数: {}" , context.readSheetHolder().getTotal()); log.info("Excel解析完成，总处理条数: {}", context.readSheetHolder().getTotal());
log.info("成功数量：{}；失败数量：{}",success.get(),fails.get()); log.info("成功数量：{}；失败数量：{}", success.get(), fails.get());
// 关闭线程池 // 关闭线程池
executorService.shutdown(); executorService.shutdown();
} }
@ -105,26 +105,26 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
private void submitBatch() { private void submitBatch() {
// 复制当前批次数据（避免异步修改） // 复制当前批次数据（避免异步修改）
List<SxGoosModelExcel> batchCopy = new ArrayList<>(deduplicateById(cachedDataList)); List<SxGoosModelExcel> batchCopy = new ArrayList<>(deduplicateById(cachedDataList));
log.info("去重前：{};去重后：{}" , cachedDataList.size(), batchCopy.size()); log.info("去重前：{};去重后：{}", cachedDataList.size(), batchCopy.size());
final int index = batchSize.get(); final int index = batchSize.get();
futures.add(executorService.submit(()->{ futures.add(executorService.submit(() -> {
int i=0; int i = 0;
while (true){ while (true) {
i++; i++;
try { try {
Gson gson=new Gson(); Gson gson = new Gson();
String jsonShops=gson.toJson(batchCopy); String jsonShops = gson.toJson(batchCopy);
JSONArray jsonArray=new JSONArray(jsonShops); JSONArray jsonArray = new JSONArray(jsonShops);
syncThirdDataService.baseSaveOrUpdateGoodsBatch(jsonArray,storeId,isNegativeAllowed,brandMaps); syncThirdDataService.baseSaveOrUpdateGoodsBatch(jsonArray, storeId, isNegativeAllowed, brandMaps);
log.info("已提交批次: {} 条", batchCopy.size()); log.info("已提交批次: {} 条", batchCopy.size());
success.getAndIncrement(); success.getAndIncrement();
return "完成批次："+index; return "完成批次：" + index;
} catch (Exception e) { } catch (Exception e) {
if(i<2){ if (i < 2) {
continue; continue;
} }
fails.getAndIncrement(); fails.getAndIncrement();
return "失败批次："+index+";失败原因:"+e.getMessage(); return "失败批次：" + index + ";失败原因:" + e.getMessage();
} }
} }
})); }));
@ -132,25 +132,26 @@ public class ShopBatchSubmitListener extends AnalysisEventListener<SxGoosModelEx
/** /**
* 数据处理 * 数据处理
*
* @param list * @param list
* @return * @return
*/ */
private List<SxGoosModelExcel> deduplicateById(List<SxGoosModelExcel> list) { private List<SxGoosModelExcel> deduplicateById(List<SxGoosModelExcel> list) {
return list.stream() return list.stream()
.peek(sxGoosModelExcel -> { .peek(sxGoosModelExcel -> {
if(StringUtils.isNotEmpty(sxGoosModelExcel.getShop_spec())){ if (StringUtils.isNotEmpty(sxGoosModelExcel.getShop_spec())) {
sxGoosModelExcel.setProduct_spec(Collections.singletonList(sxGoosModelExcel.getShop_spec())); sxGoosModelExcel.setProduct_spec(Collections.singletonList(sxGoosModelExcel.getShop_spec()));
} }
if(null==sxGoosModelExcel.getUnit()){ if (null == sxGoosModelExcel.getUnit()) {
sxGoosModelExcel.setUnit(""); sxGoosModelExcel.setUnit("");
} }
if(ObjectUtil.isEmpty(sxGoosModelExcel.getBuy_limit())){ if (ObjectUtil.isEmpty(sxGoosModelExcel.getBuy_limit())) {
sxGoosModelExcel.setBuy_limit(0); sxGoosModelExcel.setBuy_limit(0);
} }
if(StringUtils.isEmpty(sxGoosModelExcel.getBrand_name())){ if (StringUtils.isEmpty(sxGoosModelExcel.getBrand_name())) {
sxGoosModelExcel.setBrand_name("其它品牌"); sxGoosModelExcel.setBrand_name("其它品牌");
} }
if(ObjectUtil.isEmpty(sxGoosModelExcel.getStock())){ if (ObjectUtil.isEmpty(sxGoosModelExcel.getStock())) {
sxGoosModelExcel.setStock(BigDecimal.ZERO); sxGoosModelExcel.setStock(BigDecimal.ZERO);
} }
sxGoosModelExcel.setProduct_number(sxGoosModelExcel.getProduct_barcode()); sxGoosModelExcel.setProduct_number(sxGoosModelExcel.getProduct_barcode());

6
pom.xml
View File

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