ZH
FrançaisPortuguêsEnglishEspañolDeutschTiếng Việt한국어日本語简体中文
首页 / 架构与设计 / 真实案例

设计实时交易欺诈检测系统

Cris The Dev · 2026/10/12 · 11 分钟阅读 · 86

1. 概述

每当客户刷卡时，支付系统有几十毫秒的时间来回答一个问题：这笔交易是否可信？回答慢就会导致支付体验差。回答错误就会导致你失钱，或者误拒真实客户。

本文介绍如何设计一个实时欺诈检测系统。范围包括需求、数据模型、API和处理流程。文中还处理三个难点：热键上的竞争、重试时的一致性，以及处理流量突增。读完本文后，你将拥有一套完整的架构框架，可以用Spring Boot 3、Redis、PostgreSQL和Kafka来自己实现。

2. 需求

2.1. 功能需求

  • 接收一笔交易并返回决策APPROVE、REVIEW或DECLINE，附带风险分数和原因。
  • 应用规则：异常金额、高交易频率（velocity）、风险国家。
  • 保存所有决策用于审计和后续模型训练。
  • 将REVIEW决策推送给调查部门（案件管理）。

2.2. 规模和约束

指标 数值
平均负载 3,000 TPS
峰值负载（黑五、双十一） 30,000 TPS
API延迟 p99 < 50ms
决策存储 峰值时~2.6亿条记录/天，保留2年

有三个关键约束：

  1. 幂等性：支付网关可能重试同一笔交易。每次重试都必须获得相同的旧决策。
  2. 不丢失事件：返回的决策必须被记录和发送。
  3. 可解释性：每个DECLINE都必须有具体的原因列表。

3. 架构和处理流程

流程分为两部分。同步部分在支付路径上，必须快速。异步部分处理其余工作。

Payment Gateway
      │ POST /api/v1/fraud/score
      ▼
┌──────────────────────┐   velocity, profile   ┌─────────┐
│  Fraud Scoring API   │ ────────────────────▶ │  Redis  │
│  (无状态，N个pod)      │                       └─────────┘
└──────────┬───────────┘
           │ 1 transaction: 决策 + outbox
           ▼
     ┌────────────┐   outbox中继   ┌──────────────────────┐
     │ PostgreSQL │ ───────────────▶ │ Kafka: fraud.decisions│
     └────────────┘                  └──────────┬───────────┘
                                                ▼
                         案件管理、分析、ML特征存储

具体步骤：

  1. API接收请求，检查是否已有该transactionId的决策。
  2. 运行规则。需要状态（velocity）的规则以单个原子命令读写Redis。
  3. 累加分数并根据阈值做出决策。
  4. 在同一个PostgreSQL事务中记录决策和outbox记录。
  5. 一个relay读取outbox然后发布到Kafka。下游consumer处理缓慢也不会影响API。

4. 数据模型

你需要以下依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <scope>runtime</scope>
</dependency>

PostgreSQL Schema：

CREATE TABLE fraud_decision (
    transaction_id  VARCHAR(64)  PRIMARY KEY,
    card_id         VARCHAR(64)  NOT NULL,
    amount          NUMERIC(18,2) NOT NULL,
    decision        VARCHAR(10)  NOT NULL,
    score           INT          NOT NULL,
    reasons         TEXT[]       NOT NULL,
    created_at      TIMESTAMPTZ  NOT NULL DEFAULT now()
) PARTITION BY RANGE (created_at);

CREATE TABLE fraud_outbox (
    id          BIGSERIAL PRIMARY KEY,
    aggregate_key VARCHAR(64) NOT NULL,   -- card_id, 用作Kafka key
    payload     JSONB       NOT NULL,
    published   BOOLEAN     NOT NULL DEFAULT false
);
CREATE INDEX idx_outbox_unpublished ON fraud_outbox (id) WHERE NOT published;

表按日期分区。这样的话，删除旧数据只需要DROP分区，而不是DELETE数百万行。

注意： 对于分区表，主键必须包含分区列。实际上你会使用PRIMARY KEY (transaction_id, created_at)。如果想快速按单独的transaction_id查找幂等性，保持一个小的带TTL的idempotency_key表。这里简化以便阅读。

热状态存储在Redis中：

velocity:card:{cardId}     ZSET  member=transactionId, score=epochMillis   TTL 10分钟
profile:card:{cardId}      HASH  avgAmount, homeCountry                     TTL 30天

5. API

请求、响应和共享数据类型：

public record TransactionRequest(
        String transactionId, String cardId, String merchantId,
        BigDecimal amount, String currency, String country, Instant occurredAt) {}

public enum Decision { APPROVE, REVIEW, DECLINE }

public record RuleHit(String code, int score) {
    public static RuleHit none() { return new RuleHit(null, 0); }
    public boolean matched() { return code != null; }
}

public record FraudResult(String transactionId, Decision decision, int score, List<String> reasons) {}

Controller只充当转发的角色：

@RestController
@RequestMapping("/api/v1/fraud")
public class FraudController {

    private final FraudScoringService scoringService;

    public FraudController(FraudScoringService scoringService) {
        this.scoringService = scoringService;
    }

    @PostMapping("/score")
    public FraudResult score(@RequestBody TransactionRequest request) {
        return scoringService.score(request);
    }
}

一个示例调用和结果：

curl -X POST localhost:8080/api/v1/fraud/score -H 'Content-Type: application/json' -d '{
  "transactionId":"tx-1001","cardId":"card-42","merchantId":"m-7",
  "amount":15000000,"currency":"VND","country":"NG","occurredAt":"2026-10-12T08:00:00Z"}'
{"transactionId":"tx-1001","decision":"REVIEW","score":55,"reasons":["HIGH_RISK_COUNTRY","AMOUNT_SPIKE"]}

6. 规则引擎

每个规则是一个独立的bean。添加新规则不会触及旧代码：

public interface FraudRule {
    RuleHit evaluate(TransactionRequest tx);
}

@Component
public class HighRiskCountryRule implements FraudRule {
    private static final Set<String> RISKY = Set.of("NG", "KP", "IR");

    @Override
    public RuleHit evaluate(TransactionRequest tx) {
        return RISKY.contains(tx.country()) ? new RuleHit("HIGH_RISK_COUNTRY", 30) : RuleHit.none();
    }
}

服务累加分数、做出决策和保存结果：

@Service
public class FraudScoringService {

    private final List<FraudRule> rules;
    private final DecisionRepository repository;

    public FraudScoringService(List<FraudRule> rules, DecisionRepository repository) {
        this.rules = rules;
        this.repository = repository;
    }

    public FraudResult score(TransactionRequest tx) {
        Optional<FraudResult> existing = repository.find(tx.transactionId());
        if (existing.isPresent()) {
            return existing.get();               // retry → 返回旧决策
        }
        List<RuleHit> hits = rules.stream().map(r -> r.evaluate(tx)).filter(RuleHit::matched).toList();
        int score = Math.min(100, hits.stream().mapToInt(RuleHit::score).sum());
        Decision decision = score >= 80 ? Decision.DECLINE : score >= 40 ? Decision.REVIEW : Decision.APPROVE;
        FraudResult result = new FraudResult(tx.transactionId(), decision, score,
                hits.stream().map(RuleHit::code).toList());
        return repository.saveOrGetExisting(tx, result);
    }
}

Spring自动将所有FraudRule bean注入到List<FraudRule>。分数上限为100。阈值40和80应该作为配置，让风险团队自己调整，无需重新部署。

7. 难点及解决方案

7.1. 竞争：热键上的velocity检查

规则“一张卡60秒内超过5笔交易”听起来很简单。诈骗者喜欢并行发送大量请求。如果你先读数然后再写，两个请求都会读到4并都被批准。这是经典的竞态条件。

解决方案是将“删除旧的、添加新的、计数”合并到一个Lua脚本中。Redis以原子方式运行此脚本：

@Component
public class VelocityRule implements FraudRule {

    private static final RedisScript<Long> SCRIPT = RedisScript.of("""
            redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, tonumber(ARGV[1]) - tonumber(ARGV[2]))
            redis.call('ZADD', KEYS[1], ARGV[1], ARGV[3])
            redis.call('PEXPIRE', KEYS[1], ARGV[2])
            return redis.call('ZCARD', KEYS[1])
            """, Long.class);

    private final StringRedisTemplate redis;

    public VelocityRule(StringRedisTemplate redis) { this.redis = redis; }

    @Override
    public RuleHit evaluate(TransactionRequest tx) {
        try {
            Long count = redis.execute(SCRIPT, List.of("velocity:card:" + tx.cardId()),
                    String.valueOf(tx.occurredAt().toEpochMilli()), "60000", tx.transactionId());
            return count != null && count > 5 ? new RuleHit("VELOCITY_EXCEEDED", 50) : RuleHit.none();
        } catch (DataAccessException e) {
            return new RuleHit("VELOCITY_UNAVAILABLE", 20);  // 降级，不让API崩溃
        }
    }
}

有三个值得注意的细节：

  • Member是transactionId。当网关重试时，ZADD只更新旧member的分数，不增加计数。因此规则本身是幂等的。
  • 使用occurredAt而不是服务器时间。当各个pod的时钟偏差时结果不会偏离。
  • Redis Cluster按key分散。每张卡是一个独立的key，所以负载自动分散。真正的热键通常是按merchant（velocity:merchant:*）。对于这种情况，按时间桶分割key，或在本地内存中近似计数然后定期同步。

7.2. 一致性：幂等性和outbox

两个常见的一致性错误：

  1. 两个相同transactionId的请求同时到达。两者都找不到旧决策。
  2. DB写入成功但Kafka发布失败，或反之。

错误(1)通过唯一约束和ON CONFLICT解决。错误(2)通过同一事务中的outbox解决：

@Repository
public class JdbcDecisionRepository implements DecisionRepository {

    private final JdbcTemplate jdbc;
    private final ObjectMapper mapper;

    public JdbcDecisionRepository(JdbcTemplate jdbc, ObjectMapper mapper) {
        this.jdbc = jdbc;
        this.mapper = mapper;
    }

    @Override
    @Transactional
    public FraudResult saveOrGetExisting(TransactionRequest tx, FraudResult result) {
        int inserted = jdbc.update("""
                INSERT INTO fraud_decision (transaction_id, card_id, amount, decision, score, reasons)
                VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (transaction_id) DO NOTHING
                """, tx.transactionId(), tx.cardId(), tx.amount(), result.decision().name(),
                result.score(), result.reasons().toArray(String[]::new));
        if (inserted == 0) {
            return find(tx.transactionId()).orElseThrow();   // 另一个请求赢了
        }
        jdbc.update("INSERT INTO fraud_outbox (aggregate_key, payload) VALUES (?, ?::jsonb)",
                tx.cardId(), toJson(result));
        return result;
    }

    // find(...) 和 toJson(...) 省略以简洁
}

inserted == 0意味着一个并行请求已经先写入。此时你返回那个请求的决策。因此，两次调用总是获得相同的答案。

Relay按批读取outbox并以cardId作为key发布。一张卡的所有事件进入同一分区，因此保持正确的顺序：

@Scheduled(fixedDelay = 200)
@Transactional
public void relay() {
    List<Map<String, Object>> rows = jdbc.queryForList("""
            SELECT id, aggregate_key, payload::text AS payload FROM fraud_outbox
            WHERE NOT published ORDER BY id LIMIT 500 FOR UPDATE SKIP LOCKED""");
    for (Map<String, Object> row : rows) {
        kafka.send("fraud.decisions", (String) row.get("aggregate_key"), (String) row.get("payload")).join();
        jdbc.update("UPDATE fraud_outbox SET published = true WHERE id = ?", row.get("id"));
    }
}

FOR UPDATE SKIP LOCKED允许多个实例并行运行relay而不会互相干扰。这种机制保证至少一次（at-least-once）。因此下游consumer也必须是幂等的，比如在创建调查案例时使用transactionId作为键。

7.3. 流量突增

峰值负载是平均值的10倍。有三种技术可以帮你度过这一难关：

严格的时间预算。 为Redis和DB设置短超时，使缓慢的组件不会将整个API拉过50ms：

spring.data.redis.timeout=15ms
spring.datasource.hikari.connection-timeout=20
spring.datasource.hikari.maximum-pool-size=30
spring.threads.virtual.enabled=true

Java 21的虚拟线程帮助保持数千个等待I/O的请求，而不会耗尽Tomcat的线程池。

明确的降级策略。 当Redis出错时，VelocityRule返回VELOCITY_UNAVAILABLE及20分。这是一个商业选择，不是技术选择。故障开放（放行）保护收入。故障关闭（拒绝）保护安全。与风险团队事先确定此策略，并为每次降级记录指标。

将重工作推出热路径。 复杂的ML模型、调用第三方服务或按天计算特征都在Kafka consumer上运行。Kafka吸收流量激增：consumer滞后几分钟也不会影响支付。从中得到的结果被写回profile:card:*以供API在后续交易中使用。

8. 测试主要行为

以下测试检查决策逻辑和幂等性。测试使用模拟规则和内存中的repository，不需要Redis或DB：

class FraudScoringServiceTest {

    static class InMemoryRepo implements DecisionRepository {
        final Map<String, FraudResult> store = new ConcurrentHashMap<>();
        public Optional<FraudResult> find(String id) { return Optional.ofNullable(store.get(id)); }
        public FraudResult saveOrGetExisting(TransactionRequest tx, FraudResult r) {
            FraudResult prev = store.putIfAbsent(tx.transactionId(), r);
            return prev != null ? prev : r;
        }
    }

    private TransactionRequest tx(String id) {
        return new TransactionRequest(id, "card-42", "m-7", new BigDecimal("15000000"),
                "VND", "NG", Instant.parse("2026-10-12T08:00:00Z"));
    }

    @Test
    void declinesWhenScoreReachesThreshold() {
        List<FraudRule> rules = List.of(t -> new RuleHit("HIGH_RISK_COUNTRY", 30),
                                        t -> new RuleHit("VELOCITY_EXCEEDED", 50));
        var service = new FraudScoringService(rules, new InMemoryRepo());

        FraudResult result = service.score(tx("tx-1"));

        assertThat(result.decision()).isEqualTo(Decision.DECLINE);
        assertThat(result.score()).isEqualTo(80);
        assertThat(result.reasons()).containsExactly("HIGH_RISK_COUNTRY", "VELOCITY_EXCEEDED");
    }

    @Test
    void retryReturnsSameDecisionEvenIfRulesChange() {
        AtomicInteger calls = new AtomicInteger();
        FraudRule flaky = t -> calls.incrementAndGet() == 1 ? RuleHit.none() : new RuleHit("X", 90);
        var service = new FraudScoringService(List.of(flaky), new InMemoryRepo());

        FraudResult first = service.score(tx("tx-2"));
        FraudResult retry = service.score(tx("tx-2"));

        assertThat(first.decision()).isEqualTo(Decision.APPROVE);
        assertThat(retry).isEqualTo(first);
    }
}

第二个测试非常重要。即使外部状态在两次调用之间改变，重试也不应该将APPROVE改成DECLINE。使用Lua脚本时，你应该添加一个integration测试，用Testcontainers给同一张卡并行发送20个请求，并检查只有第6个及以后的请求被标记为VELOCITY_EXCEEDED。

9. 常见错误和何时不应该这样做

  • 在支付路径上同步调用ML模型而没有超时和后备。模型慢2秒一次就会导致整个支付系统都慢。
  • 使用服务器时间进行velocity窗口。pod时钟相差几秒就会导致计数错误。
  • 直接在service中发布Kafka而不是通过outbox。你会有“幽灵”决策：已返回给网关但下游没人知道。
  • 以读取然后写入的方式计数velocity（GET然后INCR）。诈骗者只需并行发送就可以绕过。
  • 将Redis视为事实来源。Redis只保持热状态并且可能丢失。事实来源是PostgreSQL和Kafka。你必须能够从中重建profile。

注意： 如果你的负载只有几十TPS且没有延迟承诺，这个架构就是过度设计。一个读写PostgreSQL的service加上几个按card_id、created_at有索引的COUNT(*)查询就足够了。只有当数据显示你真正需要时才添加Redis、outbox和Kafka。

10. 结论

文章根据分离热路径和冷路径的原则设计了一个实时欺诈检测系统。同步API只运行快速规则，使用原子Lua脚本读写Redis，并在一个事务中记录决策和outbox。所有较重的工作都通过Kafka推送。

三个难点有三个具体的解决方案。竞争通过原子操作和按卡分键来处理。一致性基于ON CONFLICT加事务型outbox。流量突增通过严格超时、明确的降级策略和吸收负载的队列来控制。从这个框架，你可以添加新规则、在consumer端插入ML模型，然后将模型分数写回Redis供API在后续交易中使用。

system-designspring-bootrediskafkafraud-detection
作者的应用AlterTurn photos into postage stamps, collected in a stamp bookiOS · Android

同主题文章