第24章 Spring Cloud开发#

1
                               query
2
                   ┌───────────────────────────┐
3
                   │                           │
4
                   │                           ▼
5
┌─────────┐   ┌─────────┐   ┌─────────┐   ┌─────────┐
6
│ Client  │──▶│   API   │──▶│Sequencer│──▶│ Engine  │
7
└─────────┘   └─────────┘   └─────────┘   └─────────┘
8
                   ▲                           │
9
                   │                           │
10
┌─────────┐   ┌─────────┐                      │
11
│ Browser │──▶│   UI    │                      │
12
└─────────┘   └─────────┘                      │
13
     ▲                                         ▼
14
     │        ┌─────────┐   ┌─────────┐   ┌─────────┐
15
     └────────│WebSocket│◀──│  Push   │◀──│Quotation│
16
              └─────────┘   └─────────┘   └─────────┘

1
  ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
2
     ┌─────────┐    ┌─────────┐
3
──┼─▶│  Order  │───▶│  Match  │ │
4
     └─────────┘    └─────────┘
5
  │       │              │      │
6
          │              │
7
  │       ▼              ▼      │
8
     ┌─────────┐    ┌─────────┐
9
  │  │  Asset  │◀───│Clearing │ │
10
     └─────────┘    └─────────┘
11
  └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘

1
version: "3"
2
services:
3
  zookeeper:
4
    image: bitnami/zookeeper:3.5
5
    container_name: zookeeper
6
    ports:
7
      - "2181:2181"
8
    environment:
9
      - ALLOW_ANONYMOUS_LOGIN=yes
10
    volumes:
11
      - "./docker/zookeeper-data:/bitnami"
12

13
  kafka:
14
    image: bitnami/kafka:3.0
15
    container_name: kafka
16
    ports:
17
      - "9092:9092"
18
    depends_on:
19
      - zookeeper
20
    environment:
21
      - KAFKA_BROKER_ID=1
22
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
23
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
24
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
25
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
26
      - ALLOW_PLAINTEXT_LISTENER=yes
27
    volumes:
28
      - "./docker/kafka-data:/bitnami"
29

30
  redis:
31
    image: redis:6.2
32
    container_name: redis
33
    ports:
34
      - "6379:6379"
35
    volumes:
36
      - "./docker/redis-data:/data"
37

38
  mysql:
39
    image: mysql:8.0
40
    container_name: mysql
41
    ports:
42
      - "3306:3306"
43
    command: --default-authentication-plugin=mysql_native_password
44
    environment:
45
      - MYSQL_ROOT_PASSWORD=password
46
    volumes:
47
      - "./sql/schema.sql:/docker-entrypoint-initdb.d/1-schema.sql:ro"
48
      - "./docker/mysql-data:/var/lib/mysql"

1
build $ docker-compose up -d
2
Creating network "build_default" with the default driver
3
Creating zookeeper ... done
4
Creating mysql     ... done
5
Creating redis     ... done
6
Creating kafka     ... done

1
## application.yml
2
## 通用配置:
3
spring:
4
  datasource:
5
    url: jdbc:mysql://localhost/test
6

7
---
8

9
## test profile:
10
spring:
11
  config:
12
    activate:
13
      on-profile: test
14
  datasource:
15
    url: jdbc:mysql://172.16.0.100/test

1
<dependency>
2
    <groupId>org.springframework.cloud</groupId>
3
    <artifactId>spring-cloud-config-server</artifactId>
4
</dependency>

1
@EnableConfigServer
2
@SpringBootApplication
3
public class ConfigApplication {
4
    public static void main(String[] args) {
5
        SpringApplication.run(ConfigApplication.class, args);
6
    }
7
}

1
## 配置服务器的端口，通常设置为8888:
2
server:
3
  port: 8888
4

5
spring:
6
  application:
7
    name: config-server
8
  profiles:
9
    # 从文件读取配置时，Config Server激活的profile必须设定为native:
10
    active: native
11
  cloud:
12
    config:
13
      server:
14
        native:
15
          # 设置配置文件的搜索路径:
16
          search-locations: file:./config-repo, file:../config-repo, file:../../config-repo

1
config-repo/
2
├── application-default.yml
3
├── application-test.yml
4
├── application.yml
5
├── push.yml
6
├── quotation.yml
7
├── trading-api.yml
8
├── trading-engine.yml
9
├── trading-sequencer.yml
10
├── ui-default.yml
11
└── ui.yml

1
spring:
2
  application:
3
    # 设置app名称:
4
    name: ui
5
  config:
6
    # 导入Config Server地址:
7
    import: configserver:${CONFIG_SERVER:http://localhost:8888}

1
<dependency>
2
    <groupId>org.springframework.cloud</groupId>
3
    <artifactId>spring-cloud-starter-config</artifactId>
4
</dependency>

1
{
2
    "name": "ui",
3
    "profiles": [
4
        "default"
5
    ],
6
    "label": null,
7
    "version": null,
8
    "state": null,
9
    "propertySources": [
10
        {
11
            "name": "file:../config-repo/ui-default.yml",
12
            "source": {...}
13
        },
14
        {
15
            "name": "file:../config-repo/application-default.yml",
16
            "source": {...}
17
        },
18
        {
19
            "name": "file:../config-repo/ui.yml",
20
            "source": {...}
21
        },
22
        {
23
            "name": "file:../config-repo/application.yml",
24
            "source": {...}
25
        }
26
    ]
27
}

1
spring:
2
  datasource:
3
    url: jdbc:mysql://localhost/test

1
spring:
2
  datasource:
3
    url: jdbc:mysql://172.16.0.100/prod_db

1
server:
2
  port: ${APP_PORT:8000}

1
$ java -DAPP_PORT=7000 -jar ui.jar

1
java.lang.IllegalStateException: Unable to load config data from 'configserver:http://localhost:8888'
2
  at org.springframework.boot.context.config.StandardConfigDataLocationResolver.getReferences
3
    at ...

1
// 用户ID -> (资产ID -> Asset)
2
ConcurrentMap<Long, ConcurrentMap<AssetEnum, Asset>> userAssets = new ConcurrentHashMap<>();

1
public class Asset {
2
    // 可用余额:
3
    BigDecimal available;
4
    // 冻结余额:
5
    BigDecimal frozen;
6

7
    public Assets() {
8
        this(BigDecimal.ZERO, BigDecimal.ZERO);
9
    }
10

11
    public Assets(BigDecimal available, BigDecimal frozen) {
12
        this.available = available;
13
        this.frozen = frozen;
14
    }
15
}

1
public enum Transfer {
2
    // 可用转可用:
3
    AVAILABLE_TO_AVAILABLE,
4
    // 可用转冻结:
5
    AVAILABLE_TO_FROZEN,
6
    // 冻结转可用:
7
    FROZEN_TO_AVAILABLE;
8
}

1
public boolean tryTransfer(Transfer type, Long fromUser, Long toUser, AssetEnum assetId, BigDecimal amount, boolean checkBalance) {
2
    // 转账金额不能为负:
3
    if (amount.signum() < 0) {
4
        throw new IllegalArgumentException("Negative amount");
5
    }
6
    // 获取源用户资产:
7
    Asset fromAsset = getAsset(fromUser, assetId);
8
    if (fromAsset == null) {
9
        // 资产不存在时初始化用户资产:
10
        fromAsset = initAssets(fromUser, assetId);
11
    }
12
    // 获取目标用户资产:
13
    Asset toAsset = getAsset(toUser, assetId);
14
    if (toAsset == null) {
15
        // 资产不存在时初始化用户资产:
16
        toAsset = initAssets(toUser, assetId);
17
    }
18
    return switch (type) {
19
    case AVAILABLE_TO_AVAILABLE -> {
20
        // 需要检查余额且余额不足:
21
        if (checkBalance && fromAsset.available.compareTo(amount) < 0) {
22
            // 转账失败:
23
            yield false;
24
        }
25
        // 源用户的可用资产减少:
26
        fromAsset.available = fromAsset.available.subtract(amount);
27
        // 目标用户的可用资产增加:
28
        toAsset.available = toAsset.available.add(amount);
29
        // 返回成功:
30
        yield true;
31
    }
32
    // 从可用转至冻结:
33
    case AVAILABLE_TO_FROZEN -> {
34
        if (checkBalance && fromAsset.available.compareTo(amount) < 0) {
35
            yield false;
36
        }
37
        fromAsset.available = fromAsset.available.subtract(amount);
38
        toAsset.frozen = toAsset.frozen.add(amount);
39
        yield true;
40
    }
41
    // 从冻结转至可用:
42
    case FROZEN_TO_AVAILABLE -> {
43
        if (checkBalance && fromAsset.frozen.compareTo(amount) < 0) {
44
            yield false;
45
        }
46
        fromAsset.frozen = fromAsset.frozen.subtract(amount);
47
        toAsset.available = toAsset.available.add(amount);
48
        yield true;
49
    }
50
    default -> {
51
        throw new IllegalArgumentException("invalid type: " + type);
52
    }
53
    };
54
}

1
public void transfer(Transfer type, Long fromUser, Long toUser, AssetEnum assetId, BigDecimal amount) {
2
    if (!tryTransfer(type, fromUser, toUser, assetId, amount, true)) {
3
        throw new RuntimeException("Transfer failed");
4
    }
5
}

1
public boolean tryFreeze(Long userId, AssetEnum assetId, BigDecimal amount) {
2
    return tryTransfer(Transfer.AVAILABLE_TO_FROZEN, userId, userId, assetId, amount, true);
3
}

1
public void unfreeze(Long userId, AssetEnum assetId, BigDecimal amount) {
2
    if (!tryTransfer(Transfer.FROZEN_TO_AVAILABLE, userId, userId, assetId, amount, true)) {
3
        throw new RuntimeException("Unfreeze failed");
4
    }
5
}

1
public class AssetServiceTest {
2
    @Test
3
    void tryTransfer() {
4
        // TODO...
5
    }
6
}

1
public class OrderService {
2
    // 引用AssetService:
3
    final AssetService assetService;
4

5
    public OrderService(@Autowired AssetService assetService) {
6
        this.assetService = assetService;
7
    }
8
}

1
public class OrderEntity {
2
    // 订单ID / 定序ID / 用户ID:
3
    public Long id;
4
    public long sequenceId;
5
    public Long userId;
6

7
    // 价格 / 方向 / 状态:
8
    public BigDecimal price;
9
    public Direction direction;
10
    public OrderStatus status;
11

12
    // 订单数量 / 未成交数量:
13
    public BigDecimal quantity;
14
    public BigDecimal unfilledQuantity;
15

16
    // 创建和更新时间:
17
    public long createdAt;
18
    public long updatedAt;
19
}

1
public class OrderService {
2
    // 跟踪所有活动订单: Order ID => OrderEntity
3
    final ConcurrentMap<Long, OrderEntity> activeOrders = new ConcurrentHashMap<>();
4

5
    // 跟踪用户活动订单: User ID => Map(Order ID => OrderEntity)
6
    final ConcurrentMap<Long, ConcurrentMap<Long, OrderEntity>> userOrders = new ConcurrentHashMap<>();

1
/**
2
 * 创建订单，失败返回null:
3
 */
4
public OrderEntity createOrder(long sequenceId, long ts, Long orderId, Long userId, Direction direction, BigDecimal price, BigDecimal quantity) {
5
    switch (direction) {
6
    case BUY -> {
7
        // 买入，需冻结USD：
8
        if (!assetService.tryFreeze(userId, AssetEnum.USD, price.multiply(quantity))) {
9
            return null;
10
        }
11
    }
12
    case SELL -> {
13
        // 卖出，需冻结BTC：
14
        if (!assetService.tryFreeze(userId, AssetEnum.BTC, quantity)) {
15
            return null;
16
        }
17
    }
18
    default -> throw new IllegalArgumentException("Invalid direction.");
19
    }
20
    // 实例化Order:
21
    OrderEntity order = new OrderEntity();
22
    order.id = orderId;
23
    order.sequenceId = sequenceId;
24
    order.userId = userId;
25
    order.direction = direction;
26
    order.price = price;
27
    order.quantity = quantity;
28
    order.unfilledQuantity = quantity;
29
    order.createdAt = order.updatedAt = ts;
30
    // 添加到ActiveOrders:
31
    this.activeOrders.put(order.id, order);
32
    // 添加到UserOrders:
33
    ConcurrentMap<Long, OrderEntity> uOrders = this.userOrders.get(userId);
34
    if (uOrders == null) {
35
        uOrders = new ConcurrentHashMap<>();
36
        this.userOrders.put(userId, uOrders);
37
    }
38
    uOrders.put(order.id, order);
39
    return order;
40
}

1
public void removeOrder(Long orderId) {
2
    // 从ActiveOrders中删除:
3
    OrderEntity removed = this.activeOrders.remove(orderId);
4
    if (removed == null) {
5
        throw new IllegalArgumentException("Order not found by orderId in active orders: " + orderId);
6
    }
7
    // 从UserOrders中删除:
8
    ConcurrentMap<Long, OrderEntity> uOrders = userOrders.get(removed.userId);
9
    if (uOrders == null) {
10
        throw new IllegalArgumentException("User orders not found by userId: " + removed.userId);
11
    }
12
    if (uOrders.remove(orderId) == null) {
13
        throw new IllegalArgumentException("Order not found by orderId in user orders: " + orderId);
14
    }
15
}

1
// 根据订单ID查询Order，不存在返回null:
2
public OrderEntity getOrder(Long orderId) {
3
    return this.activeOrders.get(orderId);
4
}
5
// 根据用户ID查询用户所有活动Order，不存在返回null:
6
public ConcurrentMap<Long, OrderEntity> getUserOrders(Long userId) {
7
    return this.userOrders.get(userId);
8
}

1
// 以record实现的OrderKey:
2
public record OrderKey(long sequenceId, BigDecimal price) {
3
}

1
public class OrderBook {
2
    public final Direction direction; // 方向
3
    public final TreeMap<OrderKey, Order> book; // 排序树
4

5
    public OrderBook(Direction direction) {
6
        this.direction = direction;
7
        this.book = new TreeMap<>(???);
8
    }
9
}

1
private static final Comparator<OrderKey> SORT_SELL = new Comparator<>() {
2
    public int compare(OrderKey o1, OrderKey o2) {
3
        // 价格低在前:
4
        int cmp = o1.price().compareTo(o2.price());
5
        // 时间早在前:
6
        return cmp == 0 ? Long.compare(o1.sequenceId(), o2.sequenceId()) : cmp;
7
    }
8
};
9

10
private static final Comparator<OrderKey> SORT_BUY = new Comparator<>() {
11
    public int compare(OrderKey o1, OrderKey o2) {
12
        // 价格高在前:
13
        int cmp = o2.price().compareTo(o1.price());
14
        // 时间早在前:
15
        return cmp == 0 ? Long.compare(o1.sequenceId(), o2.sequenceId()) : cmp;
16
    }
17
};

1
public OrderBook(Direction direction) {
2
    this.direction = direction;
3
    this.book = new TreeMap<>(direction == Direction.BUY ? SORT_BUY : SORT_SELL);
4
}

1
public OrderEntity getFirst() {
2
    return this.book.isEmpty() ? null : this.book.firstEntry().getValue();
3
}
4

5
public boolean remove(OrderEntity order) {
6
    return this.book.remove(new OrderKey(order.sequenceId, order.price)) != null;
7
}
8

9
public boolean add(OrderEntity order) {
10
    return this.book.put(new OrderKey(order.sequenceId, order.price), order) == null;
11
}

1
public class MatchEngine {
2
    public final OrderBook buyBook = new OrderBook(Direction.BUY);
3
    public final OrderBook sellBook = new OrderBook(Direction.SELL);
4
    public BigDecimal marketPrice = BigDecimal.ZERO; // 最新市场价
5
    private long sequenceId; // 上次处理的Sequence ID
6
}

1
public MatchResult processOrder(long sequenceId, OrderEntity order) {
2
    ...
3
}

1
public MatchResult processOrder(long sequenceId, OrderEntity order) {
2
    switch (order.direction) {
3
    case BUY:
4
        // 买单与sellBook匹配，最后放入buyBook:
5
        return processOrder(order, this.sellBook, this.buyBook);
6
    case SELL:
7
        // 卖单与buyBook匹配，最后放入sellBook:
8
        return processOrder(order, this.buyBook, this.sellBook);
9
    default:
10
        throw new IllegalArgumentException("Invalid direction.");
11
    }
12
}
13

14
MatchResult processOrder(long sequenceId, OrderEntity takerOrder, OrderBook makerBook, OrderBook anotherBook) {
15
    ...
16
}

1
MatchResult processOrder(long sequenceId, OrderEntity takerOrder, OrderBook makerBook, OrderBook anotherBook) {
2
    this.sequenceId = sequenceId;
3
    long ts = takerOrder.createdAt;
4
    MatchResult matchResult = new MatchResult(takerOrder);
5
    BigDecimal takerUnfilledQuantity = takerOrder.quantity;
6
    for (;;) {
7
        OrderEntity makerOrder = makerBook.getFirst();
8
        if (makerOrder == null) {
9
            // 对手盘不存在:
10
            break;
11
        }
12
        if (takerOrder.direction == Direction.BUY && takerOrder.price.compareTo(makerOrder.price) < 0) {
13
            // 买入订单价格比卖盘第一档价格低:
14
            break;
15
        } else if (takerOrder.direction == Direction.SELL && takerOrder.price.compareTo(makerOrder.price) > 0) {
16
            // 卖出订单价格比买盘第一档价格高:
17
            break;
18
        }
19
        // 以Maker价格成交:
20
        this.marketPrice = makerOrder.price;
21
        // 待成交数量为两者较小值:
22
        BigDecimal matchedQuantity = takerUnfilledQuantity.min(makerOrder.unfilledQuantity);
23
        // 成交记录:
24
        matchResult.add(makerOrder.price, matchedQuantity, makerOrder);
25
        // 更新成交后的订单数量:
26
        takerUnfilledQuantity = takerUnfilledQuantity.subtract(matchedQuantity);
27
        BigDecimal makerUnfilledQuantity = makerOrder.unfilledQuantity.subtract(matchedQuantity);
28
        // 对手盘完全成交后，从订单簿中删除:
29
        if (makerUnfilledQuantity.signum() == 0) {
30
            makerOrder.updateOrder(makerUnfilledQuantity, OrderStatus.FULLY_FILLED, ts);
31
            makerBook.remove(makerOrder);
32
        } else {
33
            // 对手盘部分成交:
34
            makerOrder.updateOrder(makerUnfilledQuantity, OrderStatus.PARTIAL_FILLED, ts);
35
        }
36
        // Taker订单完全成交后，退出循环:
37
        if (takerUnfilledQuantity.signum() == 0) {
38
            takerOrder.updateOrder(takerUnfilledQuantity, OrderStatus.FULLY_FILLED, ts);
39
            break;
40
        }
41
    }
42
    // Taker订单未完全成交时，放入订单簿:
43
    if (takerUnfilledQuantity.signum() > 0) {
44
        takerOrder.updateOrder(takerUnfilledQuantity,
45
                takerUnfilledQuantity.compareTo(takerOrder.quantity) == 0 ? OrderStatus.PENDING
46
                        : OrderStatus.PARTIAL_FILLED,
47
                ts);
48
        anotherBook.add(takerOrder);
49
    }
50
    return matchResult;
51
}

1
public class MatchResult {
2
    public final Order takerOrder;
3
    public final List<MatchDetailRecord> MatchDetails = new ArrayList<>();
4

5
    // 构造方法略
6
}

1
public record MatchDetailRecord(
2
    BigDecimal price,
3
    BigDecimal quantity,
4
    OrderEntity takerOrder,
5
    OrderEntity makerOrder) {
6
}

1
// 方向 价格 数量
2
buy  2082.34 1
3
sell 2087.6  2
4
buy  2087.8  1
5
buy  2085.01 5
6
sell 2088.02 3
7
sell 2087.60 6
8
buy  2081.11 7
9
buy  2086.0  3
10
buy  2088.33 1
11
sell 2086.54 2
12
sell 2086.55 5
13
buy  2086.55 3

1
2088.02 3
2
2087.60 6
3
2086.55 4
4
---------
5
2086.55
6
---------
7
2086.00 3
8
2085.01 5
9
2082.34 1
10
2081.11 7

1
public class ClearingService {
2
    final AssetService assetService;
3
    final OrderService orderService;
4

5
    public ClearingService(@Autowired AssetService assetService, @Autowired OrderService orderService) {
6
        this.assetService = assetService;
7
        this.orderService = orderService;
8
    }
9
}

1
public void clearMatchResult(MatchResult result) {
2
    OrderEntity taker = result.takerOrder;
3
    switch (taker.direction) {
4
    case BUY -> {
5
        // TODO
6
    }
7
    case SELL -> {
8
        // TODO
9
    }
10
    default -> throw new IllegalArgumentException("Invalid direction.");
11
    }
12
}

1
case BUY -> {
2
    // 买入时，按Maker的价格成交：
3
    for (MatchDetailRecord detail : result.matchDetails) {
4
        OrderEntity maker = detail.makerOrder();
5
        BigDecimal matched = detail.quantity();
6
        if (taker.price.compareTo(maker.price) > 0) {
7
            // 实际买入价比报价低，部分USD退回账户:
8
            BigDecimal unfreezeQuote = taker.price.subtract(maker.price).multiply(matched);
9
            assetService.unfreeze(taker.userId, AssetEnum.USD, unfreezeQuote);
10
        }
11
        // 买方USD转入卖方账户:
12
        assetService.transfer(Transfer.FROZEN_TO_AVAILABLE, taker.userId, maker.userId, AssetEnum.USD, maker.price.multiply(matched));
13
        // 卖方BTC转入买方账户:
14
        assetService.transfer(Transfer.FROZEN_TO_AVAILABLE, maker.userId, taker.userId, AssetEnum.BTC, matched);
15
        // 删除完全成交的Maker:
16
        if (maker.unfilledQuantity.signum() == 0) {
17
            orderService.removeOrder(maker.id);
18
        }
19
    }
20
    // 删除完全成交的Taker:
21
    if (taker.unfilledQuantity.signum() == 0) {
22
        orderService.removeOrder(taker.id);
23
    }
24
}

1
case SELL -> {
2
    for (MatchDetailRecord detail : result.matchDetails) {
3
        OrderEntity maker = detail.makerOrder();
4
        BigDecimal matched = detail.quantity();
5
        // 卖方BTC转入买方账户:
6
        assetService.transfer(Transfer.FROZEN_TO_AVAILABLE, taker.userId, maker.userId, AssetEnum.BTC, matched);
7
        // 买方USD转入卖方账户:
8
        assetService.transfer(Transfer.FROZEN_TO_AVAILABLE, maker.userId, taker.userId, AssetEnum.USD, maker.price.multiply(matched));
9
        // 删除完全成交的Maker:
10
        if (maker.unfilledQuantity.signum() == 0) {
11
            orderService.removeOrder(maker.id);
12
        }
13
    }
14
    // 删除完全成交的Taker:
15
    if (taker.unfilledQuantity.signum() == 0) {
16
        orderService.removeOrder(taker.id);
17
    }
18
}

1
public void clearCancelOrder(OrderEntity order) {
2
    switch (order.direction) {
3
    case BUY -> {
4
        // 解冻USD = 价格 x 未成交数量
5
        assetService.unfreeze(order.userId, AssetEnum.USD, order.price.multiply(order.unfilledQuantity));
6
    }
7
    case SELL -> {
8
        // 解冻BTC = 未成交数量
9
        assetService.unfreeze(order.userId, AssetEnum.BTC, order.unfilledQuantity);
10
    }
11
    default -> throw new IllegalArgumentException("Invalid direction.");
12
    }
13
    // 从OrderService中删除订单:
14
    orderService.removeOrder(order.id);
15
}