第4章异步化改造

第1章：异步化改造——从2秒到200毫秒的蜕变

说实话，做风控系统最怕什么？

不是规则复杂，也不是模型不准。最怕的是——用户点了一下「提交」，然后盯着转圈圈等结果。2秒，在互联网时代，足够让用户关掉页面了。

我接手这个项目时，风控接口的P99响应时间是2.1秒。业务方天天抱怨，说「你们风控比银行还慢」。嗯，这话听着扎心，但确实是我们自己的问题。

1.1 同步调用的痛点

先看看原来的架构长什么样。说白了，就是一条线走到底：

用户请求 → 风控网关 → 规则引擎 → 模型服务 → 数据库 → 返回结果

每个环节都是同步阻塞的。规则引擎要等模型服务算完，模型服务要等数据库查完。我算过一笔账：

环节	耗时（ms）	说明
规则引擎	300-500	加载规则、匹配条件
模型服务	800-1200	特征计算、模型推理
数据库查询	200-400	历史数据、黑名单
其他开销	100-200	序列化、网络传输
总计	1400-2300	平均2秒左右

你看，模型服务占了将近一半的时间。但问题是——真的所有环节都需要同步等吗？

核心洞察：风控决策中，80%的请求只需要「通过/拒绝」两个结果。剩下的20%才需要详细的风险报告。但我们的系统对所有请求一视同仁，全都走完整链路。

1.2 异步化的思路

我当时的想法很简单：把「决策」和「记录」分开。

什么意思呢？

同步路径：只做最核心的决策。规则引擎+轻量模型，200ms内给出结果。
异步路径：把详细的特征计算、模型推理、数据落库扔到消息队列里，慢慢处理。

用户只需要等200ms拿到结果。至于后面的分析报告、风险归档，那是后台的事，用户不关心。

这里我画了一张改造后的架构图，你看一眼就明白了：

1.3 消息队列选型

选型这事，我踩过坑。曾经用过RabbitMQ，结果业务量一上来，消息堆积到几百万，消费者根本来不及处理。

后来我总结了几个关键指标：

吞吐量：风控系统日均处理千万级请求，单机TPS至少得5万+
可靠性：消息丢了等于风控漏判，这是绝对不能接受的
延迟：异步不代表无限延迟，后台处理最好在5秒内完成

最终我们选了RocketMQ。为什么？

支持事务消息，保证最终一致性
自带消息轨迹，排查问题方便
社区活跃，遇到问题能找到人问

小技巧：消息队列的Topic设计要按业务维度拆分。比如「风控决策结果」「风控特征计算」「风控模型推理」各一个Topic。别把所有消息塞到一个Topic里，否则排查问题你会想哭。

1.4 代码实现要点

这里我贴一段核心代码，你看下消息生产和消费的写法：

// 生产者：同步决策后发送异步消息
public class RiskDecisionProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void sendAsyncTask(RiskContext context) {
        // 构造异步消息
        AsyncRiskMessage message = new AsyncRiskMessage();
        message.setRequestId(context.getRequestId());
        message.setUserId(context.getUserId());
        message.setFeatures(context.getBasicFeatures());
        message.setDecision(context.getDecision());
        
        // 发送到异步处理Topic
        rocketMQTemplate.asyncSend(
            "risk-async-process",
            message,
            new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步消息发送成功, requestId: {}", context.getRequestId());
                }
                
                @Override
                public void onException(Throwable e) {
                    log.error("异步消息发送失败, requestId: {}", context.getRequestId(), e);
                    // 这里要告警，消息丢了是大问题
                }
            }
        );
    }
}

// 消费者：处理详细特征计算和模型推理
@RocketMQMessageListener(
    topic = "risk-async-process",
    consumerGroup = "risk-async-consumer"
)
public class AsyncRiskConsumer implements RocketMQListener<AsyncRiskMessage> {
    
    @Override
    public void onMessage(AsyncRiskMessage message) {
        try {
            // 1. 计算完整特征
            Map<String, Object> fullFeatures = featureService.computeFullFeatures(
                message.getUserId(), 
                message.getFeatures()
            );
            
            // 2. 执行完整模型推理
            RiskResult fullResult = modelService.fullInference(fullFeatures);
            
            // 3. 落库归档
            riskArchiveService.archive(message.getRequestId(), fullResult);
            
            log.info("异步处理完成, requestId: {}", message.getRequestId());
        } catch (Exception e) {
            log.error("异步处理异常, requestId: {}", message.getRequestId(), e);
            // 重试机制，最多重试3次
            throw e;
        }
    }
}

注意：消费者一定要做好幂等性处理。消息可能重复投递，同一个requestId被处理两次，会导致数据重复。我建议用数据库的唯一索引来防重。

1.5 效果与避坑

改造完成后，我们做了压测。结果很直观：

指标	改造前	改造后	提升
P99响应时间	2.1s	198ms	10.6倍
系统吞吐量	800 TPS	4500 TPS	5.6倍
服务器成本	20台	12台	节省40%

数字看着漂亮，但过程并不顺利。我踩过几个坑，分享给你：

消息堆积：有一次双十一流量暴涨，消费者处理不过来，消息堆积到1000万。后来加了动态扩容策略，根据堆积量自动增加消费者实例。
数据不一致：同步决策和异步结果偶尔对不上。原因是特征计算用了不同的数据源。解决方案是统一特征服务，同步和异步都用同一套API。
监控缺失：刚开始没做消息延迟监控，出了问题半天才发现。后来加了Prometheus告警，消息延迟超过5秒就报警。

总结一下：异步化改造的核心不是技术，而是「取舍」。你要想清楚哪些环节可以异步，哪些必须同步。我的经验是——跟用户交互的路径必须同步，后台计算和归档全部异步。这样既保证了体验，又释放了系统压力。

嗯，这一章就聊到这儿。异步化改造看起来简单，但细节很多。下一章我们聊聊「规则引擎的冷热分离」，那个更有意思。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

第1章：异步化改造——从2秒到200毫秒的蜕变

1.1 同步调用的痛点

1.2 异步化的思路

1.3 消息队列选型

1.4 代码实现要点

1.5 效果与避坑