撮合引擎开发:缓存和MQ

原创文章,转载请注明:转载自Keegan小钢
并标明原文链接:http://keeganlee.me/post/matching/20191202
微信公众号:名称:Keegan小钢;微信号:keeganlee_me
写于2019-12-02


撮合引擎开发:开篇
撮合引擎开发:MVP版本
撮合引擎开发:数据结构设计
撮合引擎开发:对接黑箱
撮合引擎开发:解密黑箱流程
撮合引擎开发:流程的代码实现
撮合引擎开发:缓存和MQ
撮合引擎开发:日志输出
撮合引擎开发:完结篇


中间件

先来回顾下我们撮合程序项目中关于中间件的目录结构:

1
2
3
4
5
6
├── middleware               # 中间件的包
│   ├── cache # 缓存包
│   │   └── cache.go # 缓存操作
│   ├── mq # 消息队列包
│   │   └── mq.go # MQ操作
│   └── redis.go # 主要做Redis初始化操作

虽然现在只用到了 Redis 一个中间件,但设计个 middleware 包,会方便以后扩展添加其他中间件,如 Kafka 或 RocketMQ 等。

再将缓存和消息队列分包,职责上就很分明,应用时也很明确。

redis.go 就只是做初始化的连接,我们来看看代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package middleware

import (
"matching/log"

"github.com/go-redis/redis"
"github.com/spf13/viper"
)

var RedisClient *redis.Client

func Init() {
addr := viper.GetString("redis.addr")
RedisClient = redis.NewClient(&redis.Options{
Addr: addr,
Password: "", // no password set
DB: 0, // use default DB
})

_, err := RedisClient.Ping().Result()
if err != nil {
panic(err)
} else {
log.Printf("Connected to redis: %s", addr)
}
}

其中,viper 是前文说过的第三方配置库,通过 viper.GetString(“redis.addr”) 从配置文件读取出要连接的 Redis 的地址,之后就新建一个 Redis 客户端并连接上 Redis 服务器了。

缓存的设计

讲数据结构设计时,我们已经说过,使用缓存的目的主要有两个:

  1. 请求去重,避免重复提交相同订单;
  2. 恢复数据,即程序重启后能恢复所有数据。

还记得上一篇文章讲 Dispatch 的实现时,有个判断订单是否存在的逻辑吗?就是读取缓存中是否已经存在该订单,从而判别是否为重复请求或无效请求。以及,还记得 process 包的初始化?就是从缓存中恢复数据的过程。

先了解下,我们总共缓存了哪些数据:

  • 开启撮合的交易标的 symbol;

  • 这些交易标的的最新价格;

  • 所有有效的订单请求,包括下单和撤单请求。

1. 缓存symbol

开启撮合的交易标的 symbol 会有多个,且不能重复,那其实就可以保存为集合 set 类型。我将该 set 的 key 设计为 matching:symbols,之后,每有一个 symbol 开启撮合时,就可以用 Redis 的 sadd 命令将该 symbol 添加进这个集合里去了。而关闭撮合时,则需用 srem 命令将关闭撮合的 symbol 从集合中移除。读取所有 symbol 则可用 smembers 命令操作。

程序里对 symbol 的操作提供了三个函数,分别用来保存 symbol、移除 symbol 和获取所有 symbol,以下是实现的代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func SaveSymbol(symbol string) {
key := "matching:symbols"
RedisClient.SAdd(key, symbol)
}

func RemoveSymbol(symbol string) {
key := "matching:symbols"
RedisClient.SRem(key, symbol)
}

func GetSymbols() []string {
key := "matching:symbols"
return RedisClient.SMembers(key).Val()
}

2. 缓存价格

交易标的的最新价格则是每个 symbol 会有一个价格,且无需缓存历史价格,那我就直接用字符串类型来保存价格,而每个价格的 key 则包含有各自的 symbol,key 的格式设计为 matching:price:{symbol},假如要保存的 symbol = “BTCUSD”,那对应的 key 值就是 matching:price:BTCUSD,保存的 value 值就是 BTCUSD 的最新价格。

我们也同样提供了保存价格、获取价格和删除价格的三个函数,代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func SavePrice(symbol string, price decimal.Decimal) {
key := "matching:price:" + symbol
RedisClient.Set(key, price.String(), 0)
}

func GetPrice(symbol string) decimal.Decimal {
key := "matching:price:" + symbol
priceStr := RedisClient.Get(key).Val()
result, err := decimal.NewFromString(priceStr)
if err != nil {
result = decimal.Zero
}
return result
}

func RemovePrice(symbol string) {
key := "matching:price:" + symbol
RedisClient.Del(key)
}

3. 缓存订单

对订单的缓存设计则没那么简单了,需要满足两点要求:

  1. 既能缓存下单请求,也能缓存撤单请求;
  2. 订单要符合定序要求。

先说下第一点,为什么需要缓存订单?且为什么下单和撤单请求都需要缓存?

先来解答第一个问题,我们是在内存中撮合的,每个交易标的引擎里各自维护了一个交易委托账本,程序运行时,这些账本是直接保存在程序内存里的。那如果程序退出了,这些账本都被清空了。如果没有缓存,那程序重启后就无法恢复账本数据。要满足该需求,就需要缓存账本里的所有委托单。

关于第二个问题,我们来考虑这样一个场景:假如订单通道里有撤单请求在排队,而程序并没有对撤单请求做缓存,这时程序重启了,那么订单通道里的所有订单还没被引擎接收处理之前就被清空了,撤单请求也就无法恢复了。

因此,程序需要缓存好订单,且下单和撤单都需要缓存。

再来看第二个要求,为什么要符合定序?我们知道,订单通道里的订单是定序的,交易委托账本里同价格的订单也是按时间排序的,那缓存时如果不定序,程序重启后就难以保证按原有的顺序恢复订单。

那具体要怎么来设计这个订单的缓存呢?我的方案是分两类缓存,第一类保存每个独立的订单请求,包括下单和撤单;第二类分交易标的保存对应 symbol 所有订单请求的订单 ID 和 action。

第一类,我设计的 Key 格式为 matching:order:{symbol}:{orderId}:{action},symbol、orderId 和 action 则是对应订单的三个变量值。比如,某订单 symbol = “BTCUSD”,orderId = “12345”,action = “cancel”,那该订单保存到 Redis 的 Key 值就是 matching:order:BTCUSD:12345:cancel。该 Key 对应的 Value 则是保存整个订单对象,可以用 hash 类型存储。

第二类,我设计的 Key 格式为 matching:orderids:{symbol},Value 保存的是 sorted set 类型的数据,保存对应 symbol 的所有订单请求,每条记录保存的值为 {orderId}:{action},而 score 值设为对应订单的 {timestamp}。用订单时间作为 score 就可以保证定序了。还记得之前文章我们将订单时间的单位设为 100 纳秒,保证时间戳长度刚好为 16 位吗?这是因为,如果超过 16 位,那 score 将转为科学计数法表示,那将会导致数字失真。

根据这样的设计,那保存订单时的实现逻辑就如以下代码所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
func SaveOrder(order map[string]interface{}) {
symbol := order["symbol"].(string)
orderId := order["orderId"].(string)
timestamp := order["timestamp"].(float64)
action := order["action"].(string)

key := "matching:order:" + symbol + ":" + orderId + ":" + action
RedisClient.HMSet(key, order)

key = "matching:orderids:" + symbol
z := &redis.Z{
Score: timestamp,
Member: orderId + ":" + action,
}
RedisClient.ZAdd(key, z)
}

另外,还提供了 GetOrder()、UpdateOrder()、RemoveOrder()、OrderExist()、GetOrderIdsWithAction() 等函数。再给大伙看看 GetOrderIdsWithAction() 函数的实现:

1
2
3
4
func GetOrderIdsWithAction(symbol string) []string {
key := "matching:orderids:" + symbol
return RedisClient.ZRange(key, 0, -1).Val()
}

该函数得到的结果是根据 score 值排好序的,这就是我们想要的结果。理解了这个设计之后,再翻回去看看 process 包的初始化,你就会明白那些代码的逻辑了。

MQ的设计

我们选择了使用 Redis 的 Stream 数据结构来作为 MQ 输出,Stream 数据结构采用了类似 Kafka 的设计,应用起来很方便。但由于 Redis 运行于内存的特性,相比 Kafka 快速很多,这也是我选择它来作为撮合程序的输出 MQ 的主要原因。

我们只有两类 MQ,撤单结果和成交记录,发送消息的实现如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
func SendCancelResult(symbol, orderId string, ok bool) {
values := map[string]interface{}{"orderId": orderId, "ok": ok}
a := &redis.XAddArgs{
Stream: "matching:cancelresults:" + symbol,
MaxLenApprox: 1000,
Values: values,
}
RedisClient.XAdd(a)
}

func SendTrade(symbol string, trade map[string]interface{}) {
a := &redis.XAddArgs{
Stream: "matching:trades:" + symbol,
MaxLenApprox: 1000,
Values: trade,
}
RedisClient.XAdd(a)
}

其中,matching:cancelresults:{symbol} 就是撤单结果的 MQ 所属的 Key,matching:trades:{symbol} 则是成交记录的 MQ 所属的 Key。可以看到,我们还根据不同 symbol 分不同 MQ,这样还方便下游服务可以根据需要实现分布式订阅不同 symbol 的 MQ。

小结

本小节讲解了缓存和 MQ 的设计与实现,理解了这部分的设计之后,对整个撮合引擎的核心设计也基本能理解了。

最后,依然留几个思考题:是否可以不用缓存?如果不用缓存可以如何解决去重和数据恢复的问题?


扫描以下二维码即可关注公众号(公众号名称:Keegan小钢)