问题背景
fundchannel-service 服务单节点负载高。
定时任务分析
Bakong 入金定时器
BakongIncomJob
*/2 * * * * ?
主要功能
- 查询 BAKONG_INCOMING_ORDER 中 SUCCESS 状态的数据，发送消息让其它服务完成后续处理
- 查询 BAKONG_INCOMING_ORDER 中 PENDING 状态的数据，调用 BAKONG 接口获取最新状态
潜在问题
-
涉及到外部调用，较为耗时
-
查询不命中索引
select * from BAKONG_INCOMING_ORDER where lower(STATUS) = lower('PENDING') and SEND_COUNT < 1 and QUERY_TIME < sysdate and query_count < 50 and ORDER_TYPE IN ('0', '1', '2', '4', '5', '7', '10') and rownum < 20
Bakong 二维码定时器
BakongQrCodeJob
0/10 * * * * ?
主要功能
BAKONG 生成商户二维码时，可选将 qrData 和 md5 保存起来，后续根据 md5 查询 Bakong 获取 transaction_hash，再做入金操作。
潜在问题
-
7月26日左右切换过地址，切换后的地址连不上 https://api-bakong.nbc.gov.kh/v1/check_transaction_by_md5 http://172.24.21.125/v1/check_transaction_by_md5
-
交易会有推送（/api/v1/transactions/callback），这个查询可以不用
-
一个二维码多笔交易，这个接口怎么获取？
向外部渠道查询交易/退款状态
主要功能
查询处理中的数据，向对应渠道获取交易/退款状态
潜在问题
-
定时任务过多且执行频率高，有12个任务做类似的功能
-
查询没有索引，且
create_time > sysdate - 30条件不算合理 -
频繁创建销毁线程池，得不偿失
CountDownLatch countDownLatch = new CountDownLatch(payJrnList.size()); int maxThreadNum = Math.min(payJrnList.size(), 10); ExecutorService executor = Executors.newFixedThreadPool(maxThreadNum); try { payJrnList.forEach(n -> { executor.submit(() -> { try { maintainingState(n); } catch (Exception e) { } finally { countDownLatch.countDown(); } }); }); countDownLatch.await(); } catch (InterruptedException e) { log.error("查询订单异常", e); } finally { executor.shutdown(); } -
不需要查询结果的交易多次出现在任务中（主要是通过 BAKONG 向银行转账）
分析总结
现有数据分析
有交易数据的支付机构
select * from (select j.*, row_number() over (partition by PAY_ORG order by CREATE_TIME desc ) as rn from PAY_FUND_CHANNEL.PAY_JRN j where j.TRADE_STATE = 12) where rn = 1;
各个支付机构交易成功前任务查询的次数
select PAY_ORG,max(QUERIES),min(QUERIES),avg(QUERIES),STATS_MODE(QUERIES) from PAY_FUND_CHANNEL.PAY_JRN j where j.TRADE_STATE = 12 and CREATE_TIME > date '2025-01-01' group by PAY_ORG
定时任务问题总结
前面两个定时任务运行正常，主要针对查询交易/退款终态的定时任务，总结两点主要问题：
- 需要查询状态的交易记录没有做重试次数限制和重试时间退避，有查询了 250W 次的记录
- 在大表 PAY_JRN（记录1500W+） 中频繁执行没有索引的查询
此外，由于一批交易查询结果的耗时不固定，没办法较为精确的控制同一笔交易两次查询状态执行的间隔。
建议解决方案
针对问题总结中提及的两点，能确定优化的方向就是合理控制重试次数以及减少查询次数。
基于定时任务方案的优化
将待查询终态的记录单独存放
创建 PAY_JRN 记录时同步往新表 PAY_JRN_PROCESSING 插入记录，存放处理中的记录，以及查询次数等信息
处理成功后，或者重试次数达到阈值后将记录删除，保持新表较少数据量
合理重试
基于上面现有数据的分析，做如下假设：
前 10 次保持 1 秒查询一次（要注意前一次查询完成后再发起下一次查询），后续按指数退避策略，或者斐波那契额数列作为间隔时间
达到阈值（一周）后不再处理，由人工介入
代码优化
有以下两点可以优化：
- 正确使用线程池，线程池不能频繁创建销毁；
- 查询中明确返回【失败】结果，而不是非【成功】就返回【处理中】，这样能避免后续无用的重复查询
基于延迟队列方案的优化
使用延迟队列配合重试完成交易记录终态的获取，步骤如下：
交易成功发起后，发送一个待查询终态的消息到中间件，消息包括交易ID、查询次数（下次查询时间）
消费消息，查询交易终态，如果未到终态，将查询次数（下次查询时间）按规则递增后继续发送待查询终态的消息
当查询到终态时更新交易状态（或查询次数达到阈值时），不再发送消息
这样做有两点好处：
- 避免频繁查询/修改数据库，减少数据库压力
- 可以较为精确的控制两次查询的间隔（因为是在当前查询完成后才发送消息决定下次查询时间）
有以下多种延迟队列方案可选：
JDK 内置 DelayQueue
public class DelayQueueTest {
public static void main(String[] args) throws InterruptedException {
DelayQueue<DelayTask> delayQueue = new DelayQueue<>();
delayQueue.add(new DelayTask("delay-1", 1, 1));
delayQueue.add(new DelayTask("delay-2", 1, 2));
while (!delayQueue.isEmpty()) {
DelayTask task = delayQueue.take();
System.out.println("doing task...");
System.out.println(task);
if (task.count > 3) {
System.out.println(task.taskId + " task count is over");
} else {
DelayTask newTask = new DelayTask(task.taskId, task.count + 1, (int) Math.pow(2, task.count));
delayQueue.add(newTask);
}
}
}
static class DelayTask implements Delayed {
private final String taskId;
private final int count;
private final long nextTime;
public DelayTask(String taskId, int count, int delay) {
this.taskId = taskId;
this.count = count;
this.nextTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(delay);
}
@Override
public long getDelay(@NotNull TimeUnit unit) {
return unit.convert(nextTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(@NotNull Delayed o) {
return Long.compare(this.nextTime, ((DelayTask) o).nextTime);
}
@Override
public String toString() {
return "DelayTask{" +
"taskId='" + taskId + '\'' +
", count=" + count +
", nextTime=" + nextTime +
'}';
}
}
}
基于 Netty 的时间轮
public class HashedWheelTimerDemo {
public static void main(String[] args) {
TimeUnit unit = TimeUnit.SECONDS;
HashedWheelTimer timer = new HashedWheelTimer(1, unit);
timer.newTimeout(new Task(timer, "task-1", 1), 1, unit);
timer.newTimeout(new Task(timer, "task-2", 1), 3, unit);
}
static class Task implements TimerTask {
private final String taskId;
private final int count;
private final HashedWheelTimer timer;
public Task(HashedWheelTimer timer, String taskId, int count) {
this.taskId = taskId;
this.count = count;
this.timer = timer;
}
@Override
public void run(Timeout timeout) {
System.out.println("doing task..." + taskId);
if (count < 3) {
timer.newTimeout(new Task(timer, taskId, count + 1), (int) Math.pow(count, 2), TimeUnit.MINUTES);
} else {
System.out.println(taskId + " task count is over");
}
}
}
}
RocketMQ 延迟消息
https://rocketmq.apache.org/zh/docs/4.x/producer/04message3
Redisson 延迟队列
https://redisson.pro/docs/data-and-services/queues/#delayed-queue
public class RDelayedQueueTest {
public static void main(String[] args) throws InterruptedException {
Config config = new Config();
config.useSentinelServers()
.setMasterName("mymaster")
.setSentinelAddresses(Arrays.asList("redis://redis-node-0.redis-headless.cool-tool-uat.svc.cluster.local:26379",
"redis://redis-node-1.redis-headless.cool-tool-uat.svc.cluster.local:26379",
"redis://redis-node-2.redis-headless.cool-tool-uat.svc.cluster.local:26379"));
RedissonClient client = Redisson.create(config);
RBlockingQueue<Object> destQueue = client.getBlockingQueue("dest_queue");
RDelayedQueue<Object> delayedQueue = client.getDelayedQueue(destQueue);
delayedQueue.offer("test-task-1", 5, TimeUnit.SECONDS);
delayedQueue.offer("test-task-2", 8, TimeUnit.SECONDS);
delayedQueue.offer("test-task-3", 10, TimeUnit.SECONDS);
while (true) {
Object task = destQueue.take();
System.out.println(task);
}
// delayedQueue.destroy();
// client.shutdown();
}
}