修改同步计费

2026-10-13 10:21:28 +08:00
parent 26bdd3ba87
commit 4d27887d4a
14 changed files with 1231 additions and 12 deletions
@@ -25,4 +25,6 @@ public class TabOrderMonitorQuery implements Serializable
    private String orgOrderId;
    /** 转单号 TRANSFER_ORDER_ID */
    private String transferOrderId;
+    /** 分批同步计费每批条数 */
+    private Integer batchSize;
 }
@@ -12,6 +12,8 @@ public final class EsLogType
    public static final String CREDIT = "CREDIT";
    /** Clear order credit arrears after balance in */
    public static final String ARREARS = "ARREARS";
+    /** Monitor: sync billing from TAB_ORDER */
+    public static final String ORDER_SOURCE_SYNC = "ORDER_SOURCE_SYNC";

    private EsLogType()
    {
@@ -0,0 +1,45 @@
+package com.bms.price.search;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import com.alibaba.fastjson2.JSON;
+import com.bms.common.utils.SecurityUtils;
+import com.bms.common.utils.StringUtils;
+
+/** Write order-source sync actions to ES. */
+public final class TabOrderSyncEsLog
+{
+    private TabOrderSyncEsLog()
+    {
+    }
+
+    public static String operator()
+    {
+        try
+        {
+            String name = SecurityUtils.getUsername();
+            return StringUtils.isEmpty(name) ? "unknown" : name;
+        }
+        catch (Exception e)
+        {
+            return "unknown";
+        }
+    }
+
+    public static void save(IBillSearchStore store, String bizId, String action, Object request, Object response,
+        String status, String errorMessage)
+    {
+        if (store == null || StringUtils.isEmpty(bizId))
+        {
+            return;
+        }
+        Map<String, Object> req = new LinkedHashMap<String, Object>();
+        req.put("action", action);
+        req.put("operator", operator());
+        req.put("payload", request);
+        String reqJson = JSON.toJSONString(req);
+        String respJson = response == null ? null : JSON.toJSONString(response);
+        String st = StringUtils.isEmpty(status) ? EsBizLogSupport.STATUS_SUCCESS : status;
+        store.save(EsLogType.ORDER_SOURCE_SYNC, bizId, reqJson, respJson, st, errorMessage);
+    }
+}
@@ -11,6 +11,9 @@ public interface IPrcBillingTaskService
    /** Warehouse inbound: create task + publish MQ. */
    PrcBillingTask submitInbound(BillingTaskSubmitRequest request);

+    /** Monitor sync: create task then process immediately (no extra MQ). */
+    PrcBillingTask submitAndProcess(BillingTaskSubmitRequest request);
+
    /** MQ consumer entry: Oracle query -> bill -> writeback success status. */
    void processTask(Long taskId);

@@ -106,6 +106,52 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
        return task;
    }

+    @Override
+    public PrcBillingTask submitAndProcess(BillingTaskSubmitRequest request)
+    {
+        PrcBillingTask task = createWaitingTask(request, "monitor");
+        processTask(task.getTaskId());
+        return billingTaskMapper.selectPrcBillingTaskById(task.getTaskId());
+    }
+
+    private PrcBillingTask createWaitingTask(BillingTaskSubmitRequest request, String createBy)
+    {
+        if (request == null)
+        {
+            throw new ServiceException("billing task request required");
+        }
+        String transferNo = trim(request.getTransferNo());
+        String trackingNo = trim(request.getTrackingNo());
+        String orderNo = trim(request.getOrderNo());
+        Date billTime = request.getBillTime();
+        if (StringUtils.isEmpty(transferNo) && StringUtils.isEmpty(trackingNo) && StringUtils.isEmpty(orderNo))
+        {
+            throw new ServiceException("transferNo or trackingNo or orderNo required");
+        }
+        if (billTime == null)
+        {
+            throw new ServiceException("billTime required (billing time from mobile)");
+        }
+        BillTaskType type = BillTaskType.fromCode(request.getBillType());
+        if (type == null)
+        {
+            type = resolveAutoType(transferNo, trackingNo, orderNo);
+        }
+        PrcBillingTask task = new PrcBillingTask();
+        task.setTransferNo(transferNo);
+        task.setTrackingNo(trackingNo);
+        task.setOrderNo(orderNo);
+        task.setCustomerCode(trim(request.getCustomerCode()));
+        task.setBillTime(billTime);
+        task.setBillType(type.getCode());
+        task.setSuccessStatus(BillTaskSuccessStatus.WAITING.getCode());
+        task.setCreateBy(StringUtils.isEmpty(createBy) ? "tms" : createBy);
+        billingTaskMapper.insertPrcBillingTask(task);
+        log.info("billing task created taskId={} type={} billTime={} customerCode={} transferNo={} trackingNo={} orderNo={}",
+            task.getTaskId(), task.getBillType(), billTime, task.getCustomerCode(), transferNo, trackingNo, orderNo);
+        return task;
+    }
+
    private static final int BILL_DB_RETRY_MAX = 3;
    private static final long BILL_DB_RETRY_BASE_MS = 80L;

@@ -0,0 +1,185 @@
+package com.bms.price.task;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+import com.bms.common.core.redis.RedisCache;
+import com.bms.common.utils.StringUtils;
+import com.bms.price.domain.PrcBillingTask;
+import com.bms.price.domain.PrcBillingTaskRetryTask;
+import com.bms.price.domain.vo.BillingTaskSubmitRequest;
+import com.bms.price.domain.vo.TabOrderMonitorQuery;
+import com.bms.price.domain.vo.TabOrderMonitorVo;
+import com.bms.price.enums.BillTaskSuccessStatus;
+import com.bms.price.oracle.TabOrderMonitorService;
+import com.bms.price.search.IBillSearchStore;
+import com.bms.price.search.TabOrderSyncEsLog;
+import com.bms.price.service.IPrcBillingTaskService;
+
+/**
+ * Page Oracle TAB_ORDER then submitAndProcess each row.
+ */
+@Component
+public class TabOrderSyncAsyncRunner
+{
+    private static final Logger log = LoggerFactory.getLogger(TabOrderSyncAsyncRunner.class);
+
+    public static final String CACHE_KEY_PREFIX = "billing:orderSource:sync:";
+    public static final int TTL_MINUTES = 120;
+    public static final int DEFAULT_BATCH = 50;
+    public static final int MIN_BATCH = 20;
+    public static final int MAX_BATCH = 200;
+
+    @Autowired
+    private TabOrderMonitorService tabOrderMonitorService;
+
+    @Autowired
+    @Lazy
+    private IPrcBillingTaskService billingTaskService;
+
+    @Autowired
+    private RedisCache redisCache;
+
+    @Autowired
+    private IBillSearchStore billSearchStore;
+
+    @Async("threadPoolTaskExecutor")
+    public void run(String asyncTaskId, TabOrderMonitorQuery query, int batchSize)
+    {
+        int batch = clampBatch(batchSize);
+        try
+        {
+            long totalLong = tabOrderMonitorService.count(query);
+            int total = totalLong > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) totalLong;
+            put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, 0, "starting", total, 0, 0, 0, null);
+            if (total <= 0)
+            {
+                put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_SUCCESS, 100, "no tasks", 0, 0, 0, 0, null);
+                return;
+            }
+            int processed = 0;
+            int ok = 0;
+            int fail = 0;
+            int pageNum = 1;
+            while (true)
+            {
+                List<TabOrderMonitorVo> rows = tabOrderMonitorService.selectPage(query, pageNum, batch);
+                if (rows == null || rows.isEmpty())
+                {
+                    break;
+                }
+                put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING,
+                    Math.min(99, (int) ((processed * 100L) / Math.max(1, total))),
+                    "processing " + processed + "/" + total, total, processed, ok, fail, null);
+                for (TabOrderMonitorVo row : rows)
+                {
+                    try
+                    {
+                        PrcBillingTask task = billingTaskService.submitAndProcess(toSubmit(row));
+                        if (task != null && BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus()))
+                        {
+                            ok++;
+                        }
+                        else
+                        {
+                            fail++;
+                        }
+                    }
+                    catch (RuntimeException e)
+                    {
+                        fail++;
+                        log.warn("orderSource sync failed orderId={} transfer={} msg={}",
+                            row == null ? null : row.getOrderId(),
+                            row == null ? null : row.getTransferOrderId(),
+                            e.getMessage());
+                    }
+                    processed++;
+                    int percent = Math.min(99, (int) ((processed * 100L) / Math.max(1, total)));
+                    put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_RUNNING, percent,
+                        "processing " + processed + "/" + total, total, processed, ok, fail, null);
+                }
+                if (rows.size() < batch)
+                {
+                    break;
+                }
+                pageNum++;
+            }
+            put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_SUCCESS, 100,
+                "done ok=" + ok + " fail=" + fail, total, processed, ok, fail, null);
+            java.util.Map<String, Object> resp = new java.util.LinkedHashMap<String, Object>();
+            resp.put("asyncTaskId", asyncTaskId);
+            resp.put("total", Integer.valueOf(total));
+            resp.put("processed", Integer.valueOf(processed));
+            resp.put("ok", Integer.valueOf(ok));
+            resp.put("fail", Integer.valueOf(fail));
+            TabOrderSyncEsLog.save(billSearchStore, asyncTaskId, "FILTER_DONE", query, resp, "SUCCESS", null);
+            log.info("orderSource sync done asyncTaskId={} total={} ok={} fail={}", asyncTaskId, total, ok, fail);
+        }
+        catch (Exception e)
+        {
+            log.error("orderSource sync failed asyncTaskId={}", asyncTaskId, e);
+            String msg = e.getMessage();
+            if (StringUtils.isEmpty(msg))
+            {
+                msg = e.getClass().getSimpleName();
+            }
+            put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_FAIL, 100, "failed", 0, 0, 0, 0, msg);
+            TabOrderSyncEsLog.save(billSearchStore, asyncTaskId, "FILTER_DONE", query, null, "FAIL", msg);
+        }
+    }
+
+    public void put(String taskId, String status, int percent, String message,
+        int total, int processed, int ok, int fail, String error)
+    {
+        PrcBillingTaskRetryTask task = new PrcBillingTaskRetryTask();
+        task.setTaskId(taskId);
+        task.setStatus(status);
+        task.setPercent(percent);
+        task.setMessage(message);
+        task.setTotal(total);
+        task.setProcessed(processed);
+        task.setOk(ok);
+        task.setFail(fail);
+        task.setError(error);
+        redisCache.setCacheObject(CACHE_KEY_PREFIX + taskId, task, TTL_MINUTES, TimeUnit.MINUTES);
+    }
+
+    public PrcBillingTaskRetryTask get(String taskId)
+    {
+        return redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
+    }
+
+    public static int clampBatch(Integer batchSize)
+    {
+        int n = batchSize == null ? DEFAULT_BATCH : batchSize.intValue();
+        if (n < MIN_BATCH)
+        {
+            return MIN_BATCH;
+        }
+        if (n > MAX_BATCH)
+        {
+            return MAX_BATCH;
+        }
+        return n;
+    }
+
+    public static BillingTaskSubmitRequest toSubmit(TabOrderMonitorVo row)
+    {
+        if (row == null)
+        {
+            throw new com.bms.common.exception.ServiceException("order row required");
+        }
+        BillingTaskSubmitRequest req = new BillingTaskSubmitRequest();
+        req.setTransferNo(row.getTransferOrderId());
+        req.setTrackingNo(row.getOrgOrderId());
+        req.setOrderNo(row.getOrderId());
+        req.setCustomerCode(row.getCustomerCode());
+        req.setBillTime(row.getPickupDate());
+        return req;
+    }
+}