Public/cf_imes_server
6
0
Fork 0
mirror of http://192.168.1.205:9980/cf_devdept2/cf_imes_server.git synced 2026-08-14 13:42:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

1、system rabbitmq队列声明暂注释；2、cad异步拆单锁调整；

Browse Source
This commit is contained in:
gaoqr
2025-05-14 11:41:46 +08:00
parent cb568c5646
commit 47d473fc2d
5 changed files with 308 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/WebCadOrderImportService.java
+2 -1
View File
@@ -14,6 +14,7 @@ public interface WebCadOrderImportService {
* webcad生产单导入 * webcad生产单导入
* *
* @param file * @param file
* @return true：同步导入、false：异步导入
*/ */
void webCadOrderImport(MultipartFile file); boolean webCadOrderImport(MultipartFile file);
} }
cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/WebCadOrderImportServiceImpl.java
+48 -11
View File
@@ -55,6 +55,7 @@ import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.util.zip.GZIPInputStream; import java.util.zip.GZIPInputStream;
import static com.cf.imes.framework.security.core.util.SecurityFrameworkUtils.getUserOrganId;
import static com.cf.imes.module.executor.enums.ErrorCodeConstants.ORDER_IMPORT_ORGAN_LOCK_ERROR; import static com.cf.imes.module.executor.enums.ErrorCodeConstants.ORDER_IMPORT_ORGAN_LOCK_ERROR;
import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_IMPORT_FAILED; import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_IMPORT_FAILED;
import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_IMPORT_ORGANID_EMPTY_ERROR; import static com.cf.imes.module.plan.enums.ErrorCodeConstants.WEBCAD_ORDER_IMPORT_ORGANID_EMPTY_ERROR;
@@ -138,18 +139,20 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
private static final String REDIS_UNIQUEKEY = "webcad"; private static final String REDIS_UNIQUEKEY = "webcad";
@Override @Override
public void webCadOrderImport(MultipartFile file) { public boolean webCadOrderImport(MultipartFile file) {
Long organId = OrganContextHolder.getOrganId(); Long organId = OrganContextHolder.getOrganId();
if (ObjectUtil.isNull(organId)) { if (ObjectUtil.isNull(organId)) {
throw new ServiceException(WEBCAD_ORDER_IMPORT_ORGANID_EMPTY_ERROR); throw new ServiceException(WEBCAD_ORDER_IMPORT_ORGANID_EMPTY_ERROR);
} }
// 上锁
addLock(organId);
TransactionStatus transactionStatus = transactionManager.getTransaction(TransactionDefinition.withDefaults()); TransactionStatus transactionStatus = transactionManager.getTransaction(TransactionDefinition.withDefaults());
// 是否同步处理 // 是否同步处理
boolean sync = true; boolean sync = true;
// 异步发送是否成功，没有成功需要把锁解开
boolean initAsyncSuccess = false;
Long taskId = null; Long taskId = null;
JsonFactory factory = new JsonFactory(); JsonFactory factory = new JsonFactory();
try (GZIPInputStream gzipInputStream = new GZIPInputStream(file.getInputStream()); try (GZIPInputStream gzipInputStream = new GZIPInputStream(file.getInputStream());
@@ -172,6 +175,8 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
fileName = parser.getText(); fileName = parser.getText();
} else if ("blockData".equals(fieldName)) { } else if ("blockData".equals(fieldName)) {
if (plateCount < webcadAsyncPlateThreshold) { if (plateCount < webcadAsyncPlateThreshold) {
// 上锁
addSyncLock(organId);
// 在异步阈值范围内，直接解析整个对象 // 在异步阈值范围内，直接解析整个对象
ObjectMapper mapper = new ObjectMapper(); ObjectMapper mapper = new ObjectMapper();
@@ -193,8 +198,13 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
// 缓存请求，创建导入任务 // 缓存请求，创建导入任务
taskId = initiateTask(node, organId, fileName); taskId = initiateTask(node, organId, fileName);
// 上锁
addAsyncLock(taskId);
// 发送异步消息 // 发送异步消息
sendMessage(taskId, organId); sendMessage(taskId);
initAsyncSuccess = true;
// 跳过当前 blockData 解析步骤 // 跳过当前 blockData 解析步骤
parser.skipChildren(); // 忽略 blockData parser.skipChildren(); // 忽略 blockData
@@ -222,11 +232,9 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
log.error(webcadOrderImportFailed.getMsg(), e); log.error(webcadOrderImportFailed.getMsg(), e);
throw new ServiceException(webcadOrderImportFailed); throw new ServiceException(webcadOrderImportFailed);
} finally { } finally {
if (sync) { unlock(sync, initAsyncSuccess, organId, taskId);
// 导入锁解锁
redisLockUtil.unlock(String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId), REDIS_UNIQUEKEY);
}
} }
return sync;
} }
/** /**
@@ -253,7 +261,7 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
* *
* @param taskId * @param taskId
*/ */
private void sendMessage(Long taskId, Long organId) { private void sendMessage(Long taskId) {
Message message = MessageBuilder Message message = MessageBuilder
.withBody(taskId.toString().getBytes()) .withBody(taskId.toString().getBytes())
.setHeader("peek", DynamicDataSourceContextHolder.peek()) .setHeader("peek", DynamicDataSourceContextHolder.peek())
@@ -275,11 +283,11 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
} }
/** /**
* 加导入锁 * 加同步导入锁
* *
* @param organId * @param organId
*/ */
private void addLock(Long organId) { private void addSyncLock(Long organId) {
String importLockKey = String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId); String importLockKey = String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId);
// 检查机构导入锁 // 检查机构导入锁
boolean lockResult = redisLockUtil.lock(importLockKey, REDIS_UNIQUEKEY, chenfengCacheProperties.getLockTimeout()); boolean lockResult = redisLockUtil.lock(importLockKey, REDIS_UNIQUEKEY, chenfengCacheProperties.getLockTimeout());
@@ -288,6 +296,35 @@ public class WebCadOrderImportServiceImpl implements WebCadOrderImportService {
} }
} }
/**
* 加异步导入锁
*
* @param taskId
*/
private void addAsyncLock(Long taskId) {
String importLockKey = String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, getUserOrganId());
// 检查机构导入锁
boolean lockResult = redisLockUtil.lock(importLockKey, String.valueOf(taskId), chenfengCacheProperties.getLockTimeout());
if (!lockResult) {
throw new ServiceException(ORDER_IMPORT_ORGAN_LOCK_ERROR);
}
}
/**
* 解锁
*/
private void unlock(boolean sync, boolean initAsyncSuccess, Long organId, Long taskId) {
if(sync) {
// 解同步锁
redisLockUtil.unlock(String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId), REDIS_UNIQUEKEY);
} else {
// 异步发送消息失败了，解开异步锁
if (!initAsyncSuccess && ObjectUtil.isNotNull(taskId)) {
redisLockUtil.unlock(String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId), String.valueOf(taskId));
}
}
}
/** /**
* 异常时移除临时文件 * 异常时移除临时文件
cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/consumer/WebCadOrderImportConsumer.java
+1 -3
View File
@@ -104,8 +104,6 @@ public class WebCadOrderImportConsumer {
@Value("${chenfeng.plan.webcad.import.temp-file-path}") @Value("${chenfeng.plan.webcad.import.temp-file-path}")
private String webcadJsonCachePath; private String webcadJsonCachePath;
private static final String REDIS_UNIQUEKEY = "webcad";
@Resource @Resource
private DataSourceTransactionManager transactionManager; private DataSourceTransactionManager transactionManager;
@@ -169,7 +167,7 @@ public class WebCadOrderImportConsumer {
channel.basicAck(deliveryTag, false); channel.basicAck(deliveryTag, false);
} }
// 导入锁解锁 // 导入锁解锁
redisLockUtil.unlock(String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId), REDIS_UNIQUEKEY); redisLockUtil.unlock(String.format(RedisKeyConstants.ORDER_IMPORT_LOCK_KEY, organId), message);
} }
} }
cf-module-prod-plan/cf-module-prod-plan-biz/src/main/java/com/cf/imes/module/plan/service/orderImport/factory/WebCadOrderImportFactory.java
+1
View File
@@ -1497,6 +1497,7 @@ public class WebCadOrderImportFactory {
/** /**
* 从规格中获取属性，例：material、color * 从规格中获取属性，例：material、color
*
* @return * @return
*/ */
private String getFieldFromSpec(String spec, int index) { private String getFieldFromSpec(String spec, int index) {
cf-module-system/cf-module-system-biz/src/main/java/com/cf/imes/module/system/mq/consumer/advertisement/AdvertisementSendConsumer.java
+71 -71
View File
@@ -1,71 +1,71 @@
package com.cf.imes.module.system.mq.consumer.advertisement; //package com.cf.imes.module.system.mq.consumer.advertisement;
//
import com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants; //import com.cf.imes.framework.mq.rabbitmq.constant.RabbitMqConstants;
import com.cf.imes.module.system.dal.dataobject.advertisement.AdvertisementDO; //import com.cf.imes.module.system.dal.dataobject.advertisement.AdvertisementDO;
import com.cf.imes.module.system.dal.dataobject.advertisement.AdvertisementTaskDO; //import com.cf.imes.module.system.dal.dataobject.advertisement.AdvertisementTaskDO;
import com.cf.imes.module.system.dal.mysql.advertisement.AdvertisementMapper; //import com.cf.imes.module.system.dal.mysql.advertisement.AdvertisementMapper;
import com.cf.imes.module.system.dal.mysql.advertisement.AdvertisementTaskMapper; //import com.cf.imes.module.system.dal.mysql.advertisement.AdvertisementTaskMapper;
import com.cf.imes.module.system.enums.advertisement.AdvertisementStatusEnum; //import com.cf.imes.module.system.enums.advertisement.AdvertisementStatusEnum;
import com.cf.imes.module.system.enums.advertisement.MessageStatusEnum; //import com.cf.imes.module.system.enums.advertisement.MessageStatusEnum;
import com.cf.imes.module.system.mq.message.advertisement.AdvertisementSendMessage; //import com.cf.imes.module.system.mq.message.advertisement.AdvertisementSendMessage;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener; //import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
import javax.annotation.Resource; //import javax.annotation.Resource;
import java.time.LocalDateTime; //import java.time.LocalDateTime;
import java.util.Objects; //import java.util.Objects;
//
//
/** ///**
* 针对 {@link AdvertisementSendMessage} 的消费者 // * 针对 {@link AdvertisementSendMessage} 的消费者
* // *
* @author 晨丰科技 // * @author 晨丰科技
*/ // */
@Component //@Component
@Slf4j //@Slf4j
public class AdvertisementSendConsumer { //public class AdvertisementSendConsumer {
//
@Resource // @Resource
private AdvertisementMapper advertisementMapper; // private AdvertisementMapper advertisementMapper;
//
@Resource // @Resource
private AdvertisementTaskMapper advertisementTaskMapper; // private AdvertisementTaskMapper advertisementTaskMapper;
//
@RabbitListener(queues = RabbitMqConstants.SYSTEM_ADVERTISEMENT_UPDATE_DELAYED_QUEUE) // @RabbitListener(queues = RabbitMqConstants.SYSTEM_ADVERTISEMENT_UPDATE_DELAYED_QUEUE)
public void onMessage(AdvertisementSendMessage message){ // public void onMessage(AdvertisementSendMessage message){
// 1. 获取广告数据 // // 1. 获取广告数据
AdvertisementDO ad = advertisementMapper.selectById(message.getAdId()); // AdvertisementDO ad = advertisementMapper.selectById(message.getAdId());
if (ad == null || ad.getDeleted()) // if (ad == null || ad.getDeleted())
return; // return;
//
// 2. 检查消息有效性 // // 2. 检查消息有效性
AdvertisementTaskDO messageRecord = advertisementTaskMapper.selectByMessageId(message.getMessageId()); // AdvertisementTaskDO messageRecord = advertisementTaskMapper.selectByMessageId(message.getMessageId());
if (messageRecord == null || // if (messageRecord == null ||
Objects.equals(messageRecord.getMessageStatus(), MessageStatusEnum.CANCELLED.getStatus())) { // Objects.equals(messageRecord.getMessageStatus(), MessageStatusEnum.CANCELLED.getStatus())) {
return; // return;
} // }
//
// 3. 处理状态变更 // // 3. 处理状态变更
if (AdvertisementStatusEnum.PUBLISHED.getName().equals(message.getAction())) { // 待发布 // if (AdvertisementStatusEnum.PUBLISHED.getName().equals(message.getAction())) { // 待发布
if (ad.getStatus() == AdvertisementStatusEnum.UNPUBLISHED.getStatus() && // if (ad.getStatus() == AdvertisementStatusEnum.UNPUBLISHED.getStatus() &&
LocalDateTime.now().isAfter(ad.getStartTime())) { // LocalDateTime.now().isAfter(ad.getStartTime())) {
updateAdStatus(ad, messageRecord.getAdvertisementStatus()); // updateAdStatus(ad, messageRecord.getAdvertisementStatus());
} // }
} else if (AdvertisementStatusEnum.ENDED.getName().equals(message.getAction())) { // 待结束 // } else if (AdvertisementStatusEnum.ENDED.getName().equals(message.getAction())) { // 待结束
if (ad.getStatus() == AdvertisementStatusEnum.PUBLISHED.getStatus() && // if (ad.getStatus() == AdvertisementStatusEnum.PUBLISHED.getStatus() &&
LocalDateTime.now().isAfter(ad.getEndTime())) { // LocalDateTime.now().isAfter(ad.getEndTime())) {
updateAdStatus(ad, messageRecord.getAdvertisementStatus()); // updateAdStatus(ad, messageRecord.getAdvertisementStatus());
} // }
} // }
//
// 4. 标记消息为已处理 // // 4. 标记消息为已处理
messageRecord.setMessageStatus(MessageStatusEnum.PROCESSED.getStatus()); // messageRecord.setMessageStatus(MessageStatusEnum.PROCESSED.getStatus());
advertisementTaskMapper.updateMessageStateByMessageId(message.getMessageId(),false, MessageStatusEnum.PROCESSED.getStatus()); // advertisementTaskMapper.updateMessageStateByMessageId(message.getMessageId(),false, MessageStatusEnum.PROCESSED.getStatus());
} // }
//
private void updateAdStatus(AdvertisementDO ad, Integer newStatus) { // private void updateAdStatus(AdvertisementDO ad, Integer newStatus) {
advertisementMapper.updateAdStatus(ad.getId(), newStatus); // advertisementMapper.updateAdStatus(ad.getId(), newStatus);
} // }
//
} //}