订单列表重新计费优化
This commit is contained in:
wuweihua
+222
@@ -0,0 +1,222 @@
|
||||
package com.bms.price.task;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
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.config.BillingProperties;
|
||||
import com.bms.price.domain.BatchAddIncomeBillFeeTask;
|
||||
import com.bms.price.service.IPrcOrderBillService;
|
||||
|
||||
/**
|
||||
* Async local rebill from prc_order_bill only (no Oracle / no MQ).
|
||||
* Large jobs: process in pages of {@link #PAGE_SIZE}; each page runs in a short-lived worker pool.
|
||||
*/
|
||||
@Component
|
||||
public class PrcOrderBillLocalRebillAsyncRunner
|
||||
{
|
||||
private static final Logger log = LoggerFactory.getLogger(PrcOrderBillLocalRebillAsyncRunner.class);
|
||||
|
||||
public static final String CACHE_KEY_PREFIX = "price:billing:localRebill:task:";
|
||||
public static final int TTL_MINUTES = 120;
|
||||
/** Bill this many, wait for page done, then next page. */
|
||||
public static final int PAGE_SIZE = 500;
|
||||
public static final int MIN_THREADS = 2;
|
||||
public static final int MAX_THREADS = 16;
|
||||
|
||||
@Autowired
|
||||
@Lazy
|
||||
private IPrcOrderBillService orderBillService;
|
||||
|
||||
@Autowired
|
||||
private RedisCache redisCache;
|
||||
|
||||
@Autowired
|
||||
private BillingProperties billingProperties;
|
||||
|
||||
@Async("threadPoolTaskExecutor")
|
||||
public void run(String taskId, List<String> billIds)
|
||||
{
|
||||
int total = billIds == null ? 0 : billIds.size();
|
||||
AtomicInteger success = new AtomicInteger(0);
|
||||
AtomicInteger fail = new AtomicInteger(0);
|
||||
StringBuilder failTips = new StringBuilder();
|
||||
int threads = clampThreads(billingProperties == null ? null : billingProperties.getOrderSourceSyncThreads());
|
||||
ExecutorService pool = newWorkerPool(threads, "local-rebill");
|
||||
try
|
||||
{
|
||||
put(taskId, BatchAddIncomeBillFeeTask.STATUS_RUNNING, 1, "running", null, total, 0, 0);
|
||||
if (total == 0)
|
||||
{
|
||||
put(taskId, BatchAddIncomeBillFeeTask.STATUS_FAIL, 100, "failed", "no bills", 0, 0, 0);
|
||||
return;
|
||||
}
|
||||
int processed = 0;
|
||||
for (int from = 0; from < total; from += PAGE_SIZE)
|
||||
{
|
||||
int to = Math.min(from + PAGE_SIZE, total);
|
||||
List<String> page = billIds.subList(from, to);
|
||||
List<Callable<Boolean>> jobs = new ArrayList<Callable<Boolean>>(page.size());
|
||||
for (String billId : page)
|
||||
{
|
||||
final String one = billId;
|
||||
jobs.add(() -> {
|
||||
try
|
||||
{
|
||||
orderBillService.rebillLocalByBillId(one);
|
||||
success.incrementAndGet();
|
||||
return Boolean.TRUE;
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
fail.incrementAndGet();
|
||||
appendFail(failTips, one, e);
|
||||
log.warn("local rebill failed taskId={} billId={}", taskId, one, e);
|
||||
return Boolean.FALSE;
|
||||
}
|
||||
});
|
||||
}
|
||||
invokeAllQuiet(pool, jobs);
|
||||
processed = to;
|
||||
int percent = Math.min(99, (int) Math.floor(100.0 * processed / total));
|
||||
put(taskId, BatchAddIncomeBillFeeTask.STATUS_RUNNING, percent,
|
||||
"processing " + processed + "/" + total + " page=" + ((from / PAGE_SIZE) + 1),
|
||||
null, total, success.get(), fail.get());
|
||||
}
|
||||
String msg = "success=" + success.get() + ", fail=" + fail.get();
|
||||
put(taskId, BatchAddIncomeBillFeeTask.STATUS_SUCCESS, 100, msg,
|
||||
fail.get() > 0 && failTips.length() > 0 ? failTips.toString() : null,
|
||||
total, success.get(), fail.get());
|
||||
log.info("local rebill done taskId={} total={} success={} fail={} threads={} pageSize={}",
|
||||
taskId, total, success.get(), fail.get(), threads, PAGE_SIZE);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.error("local rebill aborted taskId={}", taskId, e);
|
||||
put(taskId, BatchAddIncomeBillFeeTask.STATUS_FAIL, 100, "failed", errorMessage(e),
|
||||
total, success.get(), fail.get());
|
||||
}
|
||||
finally
|
||||
{
|
||||
shutdownPool(pool);
|
||||
}
|
||||
}
|
||||
|
||||
public void put(String taskId, String status, int percent, String message, String error,
|
||||
int total, int success, int fail)
|
||||
{
|
||||
BatchAddIncomeBillFeeTask task = new BatchAddIncomeBillFeeTask();
|
||||
task.setTaskId(taskId);
|
||||
task.setStatus(status);
|
||||
task.setPercent(percent);
|
||||
task.setMessage(message);
|
||||
task.setError(error);
|
||||
task.setTotal(total);
|
||||
task.setSuccess(success);
|
||||
task.setFail(fail);
|
||||
redisCache.setCacheObject(CACHE_KEY_PREFIX + taskId, task, TTL_MINUTES, TimeUnit.MINUTES);
|
||||
}
|
||||
|
||||
public BatchAddIncomeBillFeeTask get(String taskId)
|
||||
{
|
||||
return redisCache.getCacheObject(CACHE_KEY_PREFIX + taskId);
|
||||
}
|
||||
|
||||
private static void invokeAllQuiet(ExecutorService pool, List<Callable<Boolean>> jobs) throws InterruptedException
|
||||
{
|
||||
List<Future<Boolean>> futures = pool.invokeAll(jobs);
|
||||
for (Future<Boolean> f : futures)
|
||||
{
|
||||
try
|
||||
{
|
||||
f.get();
|
||||
}
|
||||
catch (ExecutionException e)
|
||||
{
|
||||
// counted in job body
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static ExecutorService newWorkerPool(int threads, String namePrefix)
|
||||
{
|
||||
AtomicInteger seq = new AtomicInteger(0);
|
||||
ThreadFactory factory = r -> {
|
||||
Thread t = new Thread(r);
|
||||
t.setName(namePrefix + "-" + seq.incrementAndGet());
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
};
|
||||
return Executors.newFixedThreadPool(Math.max(1, threads), factory);
|
||||
}
|
||||
|
||||
private static void shutdownPool(ExecutorService pool)
|
||||
{
|
||||
if (pool == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
pool.shutdown();
|
||||
try
|
||||
{
|
||||
if (!pool.awaitTermination(60, TimeUnit.SECONDS))
|
||||
{
|
||||
pool.shutdownNow();
|
||||
}
|
||||
}
|
||||
catch (InterruptedException e)
|
||||
{
|
||||
Thread.currentThread().interrupt();
|
||||
pool.shutdownNow();
|
||||
}
|
||||
}
|
||||
|
||||
static int clampThreads(Integer n)
|
||||
{
|
||||
int v = n == null ? 8 : n.intValue();
|
||||
if (v < MIN_THREADS)
|
||||
{
|
||||
return MIN_THREADS;
|
||||
}
|
||||
if (v > MAX_THREADS)
|
||||
{
|
||||
return MAX_THREADS;
|
||||
}
|
||||
return v;
|
||||
}
|
||||
|
||||
private static void appendFail(StringBuilder failTips, String billId, Exception e)
|
||||
{
|
||||
synchronized (failTips)
|
||||
{
|
||||
if (failTips.length() >= 800)
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (failTips.length() > 0)
|
||||
{
|
||||
failTips.append("; ");
|
||||
}
|
||||
failTips.append(billId).append(":").append(errorMessage(e));
|
||||
}
|
||||
}
|
||||
|
||||
private static String errorMessage(Exception e)
|
||||
{
|
||||
return StringUtils.isEmpty(e.getMessage()) ? e.getClass().getSimpleName() : e.getMessage();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user