hy/aslan-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

结算消息批次发送报错处理

Browse Source
This commit is contained in:
tianfeng 2026-01-01 12:27:32 +08:00
parent fd59e929dc
commit 665d76fa71
1 changed files with 67 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/TeamBillTask.java
View File

@ -17,6 +17,13 @@ import java.time.LocalDateTime;
import java.time.YearMonth; import java.time.YearMonth;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -80,20 +87,68 @@ public class TeamBillTask {
* 创建本月账单，发出结算信号. * 创建本月账单，发出结算信号.
*/ */
private void consumeProcessPayOutBill() { private void consumeProcessPayOutBill() {
teamProfileService.scanStatusAvailable(teamProfiles -> { AtomicInteger total = new AtomicInteger(0);
for (TeamProfile teamProfile : teamProfiles) { AtomicInteger success = new AtomicInteger(0);
TeamBillCmd cmd = new TeamBillCmd() AtomicInteger fail = new AtomicInteger(0);
.setSysOrigin(teamProfile.getSysOrigin())
.setTeamId(teamProfile.getId())
.setRegion(teamProfile.getRegion())
.setOperationTime(TimestampUtils.now())
.setOperationBackUser(teamProfile.getCreateUser());
teamBillCycleService.createIfAbsent(cmd); ExecutorService executor = Executors.newFixedThreadPool(20);
otherMqMessage.teamBillSettle(new TeamBillSettleEvent()
.setTeamId(teamProfile.getId())); try {
teamProfileService.scanStatusAvailable(teamProfiles -> {
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (TeamProfile teamProfile : teamProfiles) {
TeamBillCmd cmd = new TeamBillCmd()
.setSysOrigin(teamProfile.getSysOrigin())
.setTeamId(teamProfile.getId())
.setRegion(teamProfile.getRegion())
.setOperationTime(TimestampUtils.now())
.setOperationBackUser(teamProfile.getCreateUser());
teamBillCycleService.createIfAbsent(cmd);
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
try {
otherMqMessage.teamBillSettle(new TeamBillSettleEvent()
.setTeamId(teamProfile.getId()));
success.incrementAndGet();
} catch (Exception e) {
fail.incrementAndGet();
log.error("发送结算消息失败: teamId={}", teamProfile.getId(), e);
}
}, executor);
futures.add(future);
total.incrementAndGet();
}
// 等待本批次完成
try {
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.get(30, TimeUnit.SECONDS);
} catch (Exception e) {
log.error("批次处理超时或异常", e);
}
log.info("批次完成: size={}, 累计成功={}, 累计失败={}",
teamProfiles.size(), success.get(), fail.get());
}, 100);
} finally {
executor.shutdown();
try {
if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
} }
}, 5000); }
log.info("结算消息发送完成: 总数={}, 成功={}, 失败={}",
total.get(), success.get(), fail.get());
} }