D6 零基础速成笔记

今天把 D1 第 11 节讲的跟单真正用代码跑起来。先把业务流程完整走一遍，后面每一节都会对应到其中某一步。

第 2–6 节讲 Kafka 怎么把"成交 → 找人 → 下单"这几步连起来；第 7 节讲"每人下多少"；第 8 节讲下单重试时最容易踩的坑。

消息队列是一个"先放着、慢慢处理"的中转站。发的一方把消息放进去就走，处理的一方按自己的速度一条条取出来处理。

Kafka 是最常用的消息队列之一，你们组用得很多。先记住三个词：topic 是"哪一类消息"，分区是"这类消息分成几条队来排"，key 决定"这条消息排哪一队"。

 topic: leader.fills   (4 partitions)

 producer ── send(key="L1") ──► hash("L1") % 4 = 2
                                         │
  P0  [ m3 ][ m7 ]                       │
  P1  [ m1 ][ m5 ][ m9 ]                 ▼
  P2  [ L1-a ][ L1-b ][ L1-c ]   ← L1 的消息全在这里，按顺序
  P3  [ m2 ][ m8 ]

图里 P0–P3 是 4 个分区；m1、m2… 是别的带单员的消息；L1-a、L1-b、L1-c 是 L1 的三条消息，全在 P2 里按顺序排着。

分区键就是发消息时拿哪个字段当 key。选哪个字段，决定了两件事：谁和谁的消息一定按顺序处理，以及流量会不会都挤进同一个分区。

 leader.fills                  ┌───────────────┐      copy.orders                ┌──────────────┐
 key = leaderId   ───────────► │ FanoutService │ ───► key = followerId  ───────► │ OrderWorkers │
                               └───────────────┘                                  └──────────────┘
 ordered per leader              1 msg ──► N msgs      spread evenly               place orders
                                 (fast: only math)     ordered per follower        in parallel

图里的英文：FanoutService = 扇出服务（第一段，只拆不下单）；OrderWorkers = 下单消费者（第二段，真正去下单）；leaderId = 带单员 ID；followerId = 跟单者 ID。"ordered per leader"= 同一个带单员的消息按顺序，"spread evenly"= 流量均匀摊开。

第一段拆消息时，要给每个跟单者算出"下多少个"。规则只有四条，但每一条都跟钱有关。

// LeaderFill（带单员成交）和 Follower（跟单者）的定义在第 10 节第 3 步
public class CopySizer {
    /** 按比例或固定金额算出原始数量，再向下取整到 lotSize */
    public static BigDecimal size(LeaderFill fill, Follower f, BigDecimal lot) {
        BigDecimal raw = f.mode() == Mode.RATIO
            ? fill.qty().multiply(f.equity()).multiply(f.value())          // 带单数量 × 跟单者本金 × 倍数
                  .divide(fill.leaderEquity(), 12, RoundingMode.DOWN)      // ÷ 带单员本金
            : f.value().divide(fill.price(), 12, RoundingMode.DOWN);       // 固定金额 ÷ 价格
        return raw.divide(lot, 0, RoundingMode.DOWN).multiply(lot);       // 除以 lot 取整数份，再乘回去 = 向下取整
    }
}

Kafka 保证消息不丢，但不保证只处理一次。同一条"带单员成交了"被处理两遍（这就叫重复消息），跟单者就会被下两次单，仓位翻倍。这是跟单系统里最危险的 bug 之一。

3000 个跟单者，总有人先下单、有人后下单。价格一直在变，先下单的人价格更好。让谁排第一，就是公平性问题。

 fill #1:  F1  F2  F3  F4  F5
 fill #2:  F2  F3  F4  F5  F1      ← 轮转：每次换一个人排第一
 fill #3:  F3  F4  F5  F1  F2

目标：发一条"带单员成交了"，看到 3 个跟单者各自收到一张订单；中途杀掉程序再启动，看到重复的订单被拒绝。

# 启动 Kafka（单机模式，端口 9092）
docker run -d --name kafka -p 9092:9092 apache/kafka:3.8.0

# 建两个 topic，各 4 个分区
docker exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic leader.fills \
  --partitions 4 --bootstrap-server localhost:9092
docker exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic copy.orders \
  --partitions 4 --bootstrap-server localhost:9092

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.8.0</version>
</dependency>

/** 带单员的一笔成交，放进 leader.fills 的就是它 */
public record LeaderFill(String fillId,         // 成交编号，比如 LF-1001
                         String leaderId,       // 带单员 ID，比如 L1
                         String symbol,         // 交易对，比如 BTC-USDT
                         String side,           // BUY / SELL
                         BigDecimal qty,        // 带单员这笔的数量
                         BigDecimal price,      // 成交价
                         BigDecimal leaderEquity) {}   // 带单员本金，按比例跟单要用

public enum Mode { RATIO, FIXED }               // 按比例 / 固定金额

/** 一个跟单者的设置 */
public record Follower(String id, BigDecimal equity, Mode mode,
                       BigDecimal value,        // RATIO 时是倍数，FIXED 时是金额 (U)
                       BigDecimal available) {} // 额度：最多能下多少 U

/** 要给某个跟单者下的一张单，放进 copy.orders 的就是它 */
public record CopyOrder(String clientOid, String followerId, String symbol, String side, BigDecimal qty) {}

// 下面几个小工具自己补上，都很短：
// toJson(obj) / parseFill(json) / parseOrder(json)：用 Jackson 的 ObjectMapper 转换
// followerRepo.findByLeader("L1")：练习里直接返回写死的 3 个 Follower
// skip(f, 原因)：打印一行"跳过 F3：太小"
static final BigDecimal LOT = new BigDecimal("0.001");       // lotSize
static final BigDecimal MIN_QTY = new BigDecimal("0.001");   // 最小下单量

Properties p = new Properties();
p.put("bootstrap.servers", "localhost:9092");
p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("acks", "all");                  // 所有同步中的副本都写成功才算发送成功（单机只有 1 个副本）
p.put("enable.idempotence", "true");   // 生产者自己重试时，不会往 Kafka 里写出重复消息

try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
    LeaderFill fill = new LeaderFill("LF-1001", "L1", "BTC-USDT", "BUY",
            new BigDecimal("1"), new BigDecimal("60000"), new BigDecimal("10000"));
    // key = 带单员 ID：同一个带单员的成交进同一个分区，按顺序处理
    producer.send(new ProducerRecord<>("leader.fills", fill.leaderId(), toJson(fill))).get();   // .get()：等 Kafka 确认收到
}

Properties c = new Properties();
c.put("bootstrap.servers", "localhost:9092");
c.put("group.id", "copy-fanout");        // 消费者组的名字：同名的消费者一起分担分区
c.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
c.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
c.put("enable.auto.commit", "false");   // 关掉自动提交：全部处理完才手动夹书签
c.put("auto.offset.reset", "earliest"); // 这个组第一次启动、还没有书签时，从最早的消息读起

// p 就是第 4 步那份生产者配置：扇出服务既是消费者（读 leader.fills），又是生产者（写 copy.orders）
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(c);
     KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
    consumer.subscribe(List.of("leader.fills"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));   // 最多等 200 毫秒，拿一批消息
        for (ConsumerRecord<String, String> r : records) {
            LeaderFill fill = parseFill(r.value());
            for (Follower f : followerRepo.findByLeader(fill.leaderId())) {
                BigDecimal qty = CopySizer.size(fill, f, LOT);       // 第 7 节的规则
                if (qty.compareTo(MIN_QTY) < 0) { skip(f, "太小"); continue; }
                if (qty.multiply(fill.price()).compareTo(f.available()) > 0) { skip(f, "额度不够"); continue; }
                String clientOid = fill.fillId() + "-" + f.id();      // 幂等编号：重试多少次都一样
                // key = 跟单者 ID：第二段按跟单者分区，流量均匀
                producer.send(new ProducerRecord<>("copy.orders", f.id(),
                        toJson(new CopyOrder(clientOid, f.id(), fill.symbol(), fill.side(), qty))));
            }
        }
        producer.flush();        // 确认这一批全部发出去了
        consumer.commitSync();   // 再夹书签。顺序反过来就可能丢单
    }
}

c.put("group.id", "copy-order-worker");  // 其余配置和第 5 步一样，只是换一个组名
Set<String> seen = new HashSet<>();   // 假装交易所：记住收到过的 clientOid

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(c)) {
    consumer.subscribe(List.of("copy.orders"));
    while (true) {
        for (ConsumerRecord<String, String> r : consumer.poll(Duration.ofMillis(200))) {
            CopyOrder o = parseOrder(r.value());
            if (!seen.add(o.clientOid())) {                 // 同一个编号第二次来：拒绝
                System.out.println("重复订单被拒绝: " + o.clientOid());
                continue;
            }
            // 真实系统在这里调交易所的下单接口；练习里打印出来就够了
            System.out.println("下单: " + o.clientOid() + " qty=" + o.qty());
        }
        consumer.commitSync();
    }
}

今天用到的词都在这里，包括前几天学过、今天又用到的。

自测 7 题