mq问题

2026-01-09 16:49:22 +08:00
parent 16a814568d
commit ef8eabb06c
5 changed files with 41 additions and 36 deletions
--- a/cash-api/order-server/src/main/java/com/czg/controller/admin/EntryManagerController.java
+++ b/cash-api/order-server/src/main/java/com/czg/controller/admin/EntryManagerController.java
@@ -36,6 +36,7 @@ public class EntryManagerController {

    @Resource
    private RabbitPublisher rabbitPublisher;
+
    /**
     * ocr识别填充
     * 阿里 ocr识别图片
@@ -86,8 +87,8 @@ public class EntryManagerController {
     * 获取进件信息
     */
    @GetMapping
-    public CzgResult<AggregateMerchantVO> getEntry(Long shopId) {
-        return CzgResult.success(shopDirectMerchantService.getEntry(shopId));
+    public CzgResult<AggregateMerchantVO> getEntry(Long shopId, String licenceNo) {
+        return CzgResult.success(shopDirectMerchantService.getEntry(shopId, licenceNo));
    }

    /**
@@ -97,8 +98,8 @@ public class EntryManagerController {
     */
    @GetMapping("queryEntry")
    @Debounce(value = "#shopId", interval = 1000 * 60 * 3)
-    public CzgResult<Boolean> queryEntry(Long shopId) {
-        entryManagerTask.entryManager(shopId);
+    public CzgResult<Boolean> queryEntry(Long shopId, String licenceNo) {
+        entryManagerTask.entryManager(shopId, licenceNo);
        return CzgResult.success();
    }

--- a/cash-api/order-server/src/main/java/com/czg/mq/EntryManagerMqListener.java
+++ b/cash-api/order-server/src/main/java/com/czg/mq/EntryManagerMqListener.java
@@ -48,23 +48,32 @@ public class EntryManagerMqListener {
    )
    @RabbitHandler
    public void handle(Message message, Channel channel, String msg) throws IOException {
-        log.info("进件1MQ对接开始shopId:{}", msg);
-        String messageId = message.getMessageProperties().getMessageId();
-        log.info("进件0MQ对接开始shopId:{}messageId:{}", msg, messageId);
-//        if (hasMessageId(messageId)) {
-//            return;
-//        }
+        log.info("进件1MQ对接开始 店铺标识:{}", msg);
+        long deliveryTag = message.getMessageProperties().getDeliveryTag();
+        if (StrUtil.isBlank(msg)) {
+            channel.basicNack(deliveryTag, false, false);
+            return;
+        }
+        String[] split = msg.split(":");
+        if (split.length != 2) {
+            log.error("进件MQ对接参数异常 店铺标识:{}", msg);
+            channel.basicNack(deliveryTag, false, false);
+            return;
+        }
+        Long shopId = Long.valueOf(split[0]);
+        if (shopId == null) {
+            channel.basicNack(deliveryTag, false, false);
+            return;
+        }
+        if (hasMessageId(String.valueOf(shopId))) {
+            return;
+        }
        try {
-            Long shopId = Long.valueOf(msg);
            // 将唯一标识添加到日志上下文
-//            ThreadContext.put("traceId", messageId);
+            ThreadContext.put("traceId", String.valueOf(shopId));
            log.info("进件2MQ对接开始shopId:{}", msg);
            // 安全转换shopId
-            if (shopId == null) {
-                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
-                return;
-            }
-            AggregateMerchantVO entry = shopDirectMerchantService.getEntry(Long.valueOf(msg));
+            AggregateMerchantVO entry = shopDirectMerchantService.getEntry(shopId, split[1]);
            log.info("进件3MQ对接开始shopId:{}", msg);
            if (entry != null) {
                EntryManager.uploadParamImage(entry);
@@ -89,7 +98,7 @@ public class EntryManagerMqListener {
                merchant.setAlipayErrorMsg(resp.getAlipayErrorMsg());
                shopDirectMerchantService.updateById(merchant);
            }
-            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("进件MQ对接业务异常shopId:{}", msg, e);
            ShopDirectMerchant merchant = new ShopDirectMerchant();
@@ -98,12 +107,11 @@ public class EntryManagerMqListener {
            merchant.setAlipayStatus(PayCst.EntryStatus.REJECTED);
            merchant.setErrorMsg("系统错误，请联系管理员后重试。");
            shopDirectMerchantService.updateById(merchant);
-
-            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
+            channel.basicNack(deliveryTag, false, false);
        } finally {
-//            delMessageId(messageId);
-            // 清除日志上下文信息
-//            ThreadContext.remove("messageId");
+            delMessageId(String.valueOf(shopId));
+//             清除日志上下文信息
+            ThreadContext.remove("messageId");
        }
    }

--- a/cash-api/order-server/src/main/java/com/czg/task/EntryManagerTask.java
+++ b/cash-api/order-server/src/main/java/com/czg/task/EntryManagerTask.java
@@ -4,14 +4,9 @@ import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.StrUtil;
 import com.czg.EntryManager;
 import com.czg.PayCst;
-import com.czg.account.entity.ShopInfo;
 import com.czg.account.service.ShopInfoService;
 import com.czg.dto.resp.QueryStatusResp;
 import com.czg.order.entity.ShopDirectMerchant;
-import com.czg.order.service.ShopOrderStatisticService;
-import com.czg.order.service.ShopProdStatisticService;
-import com.czg.order.service.ShopTableOrderStatisticService;
-import com.czg.service.RedisService;
 import com.czg.service.order.service.ShopDirectMerchantService;
 import com.mybatisflex.core.query.QueryWrapper;
 import jakarta.annotation.Resource;
@@ -20,8 +15,6 @@ import org.apache.dubbo.config.annotation.DubboReference;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;

-import java.time.LocalDate;
-import java.time.LocalDateTime;
 import java.util.List;

 /**
@@ -42,16 +35,17 @@ public class EntryManagerTask {
    public void run() {
        log.info("进件查询，定时任务执行");
        long start = System.currentTimeMillis();
-        entryManager(null);
+        entryManager(null, null);
        log.info("进件查询，定时任务执行完毕，耗时:{}ms", start - System.currentTimeMillis());
    }

    /**
     * 查询状态为待处理、待签约、待审核的进件
     */
-    public void entryManager(Long shopId) {
+    public void entryManager(Long shopId, String licenceNo) {
        List<ShopDirectMerchant> list = shopDirectMerchantService.list(QueryWrapper.create()
                .eq(ShopDirectMerchant::getShopId, shopId)
+                .eq(ShopDirectMerchant::getLicenceNo, licenceNo)
                .in(ShopDirectMerchant::getWechatStatus, PayCst.EntryStatus.NEED_QUERY_LIST)
                .or(ShopDirectMerchant::getAlipayStatus).in(PayCst.EntryStatus.NEED_QUERY_LIST));
        if (CollUtil.isEmpty(list)) {
--- a/cash-service/order-service/src/main/java/com/czg/service/order/service/ShopDirectMerchantService.java
+++ b/cash-service/order-service/src/main/java/com/czg/service/order/service/ShopDirectMerchantService.java
@@ -29,7 +29,7 @@ public interface ShopDirectMerchantService extends IService<ShopDirectMerchant>
    /**
     * 获取进件信息
     */
-    AggregateMerchantVO getEntry(Long shopId);
+    AggregateMerchantVO getEntry(Long shopId, String licenceNo);

    /**
     * 申请进件
--- a/cash-service/order-service/src/main/java/com/czg/service/order/service/impl/ShopDirectMerchantServiceImpl.java
+++ b/cash-service/order-service/src/main/java/com/czg/service/order/service/impl/ShopDirectMerchantServiceImpl.java
@@ -64,8 +64,10 @@ public class ShopDirectMerchantServiceImpl extends ServiceImpl<ShopDirectMerchan
    }

    @Override
-    public AggregateMerchantVO getEntry(Long shopId) {
-        ShopDirectMerchant merchant = getById(shopId);
+    public AggregateMerchantVO getEntry(Long shopId, String licenceNo) {
+        ShopDirectMerchant merchant = getOne(query()
+                .eq(ShopDirectMerchant::getLicenceNo, licenceNo)
+                .eq(ShopDirectMerchant::getShopId, shopId));
        if (merchant == null) {
            return null;
        }
@@ -106,7 +108,7 @@ public class ShopDirectMerchantServiceImpl extends ServiceImpl<ShopDirectMerchan
            result = updateById(merchant);
        }
        //发送进件队列消息
-        FunUtils.transactionSafeRun(() -> rabbitPublisher.sendEntryManagerMsg(reqDto.getShopId().toString()));
+        FunUtils.transactionSafeRun(() -> rabbitPublisher.sendEntryManagerMsg(reqDto.getShopId() + ":" + merchant.getLicenceNo()));
        return result;
    }