增加修改客户重新同步功能

2026-09-29 11:58:01 +08:00
parent b9746e3801
commit 2d002e7291
13 changed files with 1098 additions and 167 deletions
@@ -0,0 +1,169 @@
+package com.bms.price.task;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+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.enums.BillTaskSuccessStatus;
+import com.bms.price.mapper.PrcBillingTaskMapper;
+import com.bms.price.service.IPrcBillingTaskService;
+
+/**
+ * Async filter-based billing task retry: keyset-page IDs then retry one-by-one.
+ */
+@Component
+public class PrcBillingTaskRetryAsyncRunner
+{
+    private static final Logger log = LoggerFactory.getLogger(PrcBillingTaskRetryAsyncRunner.class);
+
+    public static final String CACHE_KEY_PREFIX = "billing:task:retry:";
+    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 PrcBillingTaskMapper billingTaskMapper;
+
+    @Autowired
+    @Lazy
+    private IPrcBillingTaskService billingTaskService;
+
+    @Autowired
+    private RedisCache redisCache;
+
+    @Async("threadPoolTaskExecutor")
+    public void run(String asyncTaskId, PrcBillingTask query, int batchSize)
+    {
+        int batch = clampBatch(batchSize);
+        try
+        {
+            int total = billingTaskMapper.countTaskIdsByFilter(query);
+            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;
+            }
+            long lastTaskId = 0L;
+            int processed = 0;
+            int ok = 0;
+            int fail = 0;
+            while (true)
+            {
+                List<Long> ids = billingTaskMapper.selectTaskIdsByFilter(query, Long.valueOf(lastTaskId), batch);
+                if (ids == null || ids.isEmpty())
+                {
+                    break;
+                }
+                for (Long taskId : ids)
+                {
+                    if (taskId == null)
+                    {
+                        continue;
+                    }
+                    lastTaskId = taskId.longValue();
+                    try
+                    {
+                        PrcBillingTask task = billingTaskService.retryTask(taskId);
+                        if (task != null && BillTaskSuccessStatus.SUCCESS.getCode().equals(task.getSuccessStatus()))
+                        {
+                            ok++;
+                        }
+                        else
+                        {
+                            fail++;
+                        }
+                    }
+                    catch (RuntimeException e)
+                    {
+                        fail++;
+                        log.warn("filter retry failed taskId={} msg={}", taskId, 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 (ids.size() < batch)
+                {
+                    break;
+                }
+            }
+            put(asyncTaskId, PrcBillingTaskRetryTask.STATUS_SUCCESS, 100,
+                "done ok=" + ok + " fail=" + fail, total, processed, ok, fail, null);
+            log.info("billing filter retry done asyncTaskId={} total={} ok={} fail={}",
+                asyncTaskId, total, ok, fail);
+        }
+        catch (Exception e)
+        {
+            log.error("billing filter retry 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);
+        }
+    }
+
+    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;
+    }
+
+    /** Build MyBatis query params map for begin/end time. */
+    public static Map<String, Object> timeParams(String beginTime, String endTime)
+    {
+        Map<String, Object> params = new HashMap<String, Object>();
+        if (StringUtils.isNotEmpty(beginTime))
+        {
+            params.put("beginTime", beginTime.trim());
+        }
+        if (StringUtils.isNotEmpty(endTime))
+        {
+            params.put("endTime", endTime.trim());
+        }
+        return params;
+    }
+}