提速修改计费重
This commit is contained in:
wuweihua
+97
-44
@@ -1,8 +1,13 @@
|
||||
package com.bms.price.task;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
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;
|
||||
@@ -15,7 +20,7 @@ import com.bms.price.domain.ChargeWeightImportTask;
|
||||
import com.bms.price.service.IPrcOrderBillService;
|
||||
|
||||
/**
|
||||
* Async runner: import charge_weight updates and rebill each order.
|
||||
* Async runner: import charge_weight updates and rebill each order (parallel).
|
||||
*/
|
||||
@Component
|
||||
public class PrcChargeWeightImportAsyncRunner
|
||||
@@ -24,6 +29,10 @@ public class PrcChargeWeightImportAsyncRunner
|
||||
|
||||
public static final String CACHE_KEY_PREFIX = "price:billing:chargeWeight:import:task:";
|
||||
public static final int TTL_MINUTES = 60;
|
||||
/** Parallel rebill workers per import task. */
|
||||
private static final int REBILL_CONCURRENCY = 8;
|
||||
/** Redis progress write every N finished rows. */
|
||||
private static final int PROGRESS_EVERY = 10;
|
||||
|
||||
@Autowired
|
||||
@Lazy
|
||||
@@ -32,28 +41,25 @@ public class PrcChargeWeightImportAsyncRunner
|
||||
@Autowired
|
||||
private RedisCache redisCache;
|
||||
|
||||
/**
|
||||
* @param items pre-validated rows with billId filled in remark field unused;
|
||||
* use orderNo + chargeWeight; billId stored in customer field as billId carrier
|
||||
* -- better use a dedicated work item. See run(List of Object[]).
|
||||
*/
|
||||
@Async("threadPoolTaskExecutor")
|
||||
public void run(String taskId, List<ChargeWeightImportWorkItem> items, int initialFail)
|
||||
{
|
||||
int total = (items == null ? 0 : items.size()) + Math.max(initialFail, 0);
|
||||
int success = 0;
|
||||
int fail = Math.max(initialFail, 0);
|
||||
int workSize = items == null ? 0 : items.size();
|
||||
int total = workSize + Math.max(initialFail, 0);
|
||||
AtomicInteger success = new AtomicInteger(0);
|
||||
AtomicInteger fail = new AtomicInteger(Math.max(initialFail, 0));
|
||||
AtomicInteger doneWork = new AtomicInteger(0);
|
||||
StringBuilder failTips = new StringBuilder();
|
||||
Object tipsLock = new Object();
|
||||
try
|
||||
{
|
||||
put(taskId, ChargeWeightImportTask.STATUS_RUNNING, 1, "running", null, total, 0, fail);
|
||||
int workSize = items == null ? 0 : items.size();
|
||||
put(taskId, ChargeWeightImportTask.STATUS_RUNNING, 1, "running", null, total, 0, fail.get());
|
||||
if (workSize == 0)
|
||||
{
|
||||
if (fail > 0)
|
||||
if (fail.get() > 0)
|
||||
{
|
||||
put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100,
|
||||
"success=0, fail=" + fail, "all rows failed validation", total, 0, fail);
|
||||
"success=0, fail=" + fail.get(), "all rows failed validation", total, 0, fail.get());
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -61,55 +67,102 @@ public class PrcChargeWeightImportAsyncRunner
|
||||
}
|
||||
return;
|
||||
}
|
||||
for (int i = 0; i < workSize; i++)
|
||||
|
||||
int poolSize = Math.min(REBILL_CONCURRENCY, workSize);
|
||||
ExecutorService pool = Executors.newFixedThreadPool(poolSize);
|
||||
List<Future<?>> futures = new ArrayList<Future<?>>(workSize);
|
||||
try
|
||||
{
|
||||
ChargeWeightImportWorkItem item = items.get(i);
|
||||
for (int i = 0; i < workSize; i++)
|
||||
{
|
||||
final ChargeWeightImportWorkItem item = items.get(i);
|
||||
futures.add(pool.submit(new Runnable()
|
||||
{
|
||||
@Override
|
||||
public void run()
|
||||
{
|
||||
try
|
||||
{
|
||||
orderBillService.updateChargeWeightByBillId(item.getBillId(), item.getChargeWeight());
|
||||
success.incrementAndGet();
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
fail.incrementAndGet();
|
||||
synchronized (tipsLock)
|
||||
{
|
||||
if (failTips.length() < 800)
|
||||
{
|
||||
String tip = e.getMessage();
|
||||
if (StringUtils.isEmpty(tip))
|
||||
{
|
||||
tip = e.getClass().getSimpleName();
|
||||
}
|
||||
if (failTips.length() > 0)
|
||||
{
|
||||
failTips.append("; ");
|
||||
}
|
||||
failTips.append(item.getOrderNo()).append(":").append(tip);
|
||||
}
|
||||
}
|
||||
log.warn("charge weight import failed taskId={} orderNo={}",
|
||||
taskId, item.getOrderNo(), e);
|
||||
}
|
||||
finally
|
||||
{
|
||||
int finished = doneWork.incrementAndGet();
|
||||
if (finished % PROGRESS_EVERY == 0 || finished == workSize)
|
||||
{
|
||||
int done = finished + Math.max(initialFail, 0);
|
||||
int percent = Math.min(99,
|
||||
(int) Math.floor(100.0 * done / Math.max(total, 1)));
|
||||
put(taskId, ChargeWeightImportTask.STATUS_RUNNING, percent,
|
||||
"processing " + done + "/" + total, null, total,
|
||||
success.get(), fail.get());
|
||||
}
|
||||
}
|
||||
}
|
||||
}));
|
||||
}
|
||||
for (Future<?> f : futures)
|
||||
{
|
||||
f.get();
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
pool.shutdown();
|
||||
try
|
||||
{
|
||||
orderBillService.updateChargeWeightByBillId(item.getBillId(), item.getChargeWeight());
|
||||
success++;
|
||||
pool.awaitTermination(30, TimeUnit.MINUTES);
|
||||
}
|
||||
catch (Exception e)
|
||||
catch (InterruptedException ie)
|
||||
{
|
||||
fail++;
|
||||
if (failTips.length() < 800)
|
||||
{
|
||||
String tip = e.getMessage();
|
||||
if (StringUtils.isEmpty(tip))
|
||||
{
|
||||
tip = e.getClass().getSimpleName();
|
||||
}
|
||||
if (failTips.length() > 0)
|
||||
{
|
||||
failTips.append("; ");
|
||||
}
|
||||
failTips.append(item.getOrderNo()).append(":").append(tip);
|
||||
}
|
||||
log.warn("charge weight import failed taskId={} orderNo={}", taskId, item.getOrderNo(), e);
|
||||
Thread.currentThread().interrupt();
|
||||
pool.shutdownNow();
|
||||
}
|
||||
int done = i + 1 + Math.max(initialFail, 0);
|
||||
int percent = Math.min(99, (int) Math.floor(100.0 * done / Math.max(total, 1)));
|
||||
put(taskId, ChargeWeightImportTask.STATUS_RUNNING, percent,
|
||||
"processing " + done + "/" + total, null, total, success, fail);
|
||||
}
|
||||
String msg = "success=" + success + ", fail=" + fail;
|
||||
if (fail > 0 && failTips.length() > 0)
|
||||
|
||||
String msg = "success=" + success.get() + ", fail=" + fail.get();
|
||||
if (fail.get() > 0 && failTips.length() > 0)
|
||||
{
|
||||
put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, msg, failTips.toString(),
|
||||
total, success, fail);
|
||||
total, success.get(), fail.get());
|
||||
}
|
||||
else
|
||||
{
|
||||
put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, msg, null, total, success, fail);
|
||||
put(taskId, ChargeWeightImportTask.STATUS_SUCCESS, 100, msg, null,
|
||||
total, success.get(), fail.get());
|
||||
}
|
||||
log.info("charge weight import done taskId={} total={} success={} fail={}",
|
||||
taskId, total, success, fail);
|
||||
log.info("charge weight import done taskId={} total={} success={} fail={} concurrency={}",
|
||||
taskId, Integer.valueOf(total), Integer.valueOf(success.get()),
|
||||
Integer.valueOf(fail.get()), Integer.valueOf(poolSize));
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.error("charge weight import aborted taskId={}", taskId, e);
|
||||
put(taskId, ChargeWeightImportTask.STATUS_FAIL, 100, "failed", errorMessage(e),
|
||||
total, success, fail);
|
||||
total, success.get(), fail.get());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user