package engine// OrderBook typetype OrderBook struct { BuyOrders []Order SellOrders []Order}// Add a buy order to the order bookfunc (book *OrderBook) addBuyOrder(order Order) { n := len(book.BuyOrders) var i int for i := n - 1; i >= 0; i-- { buyOrder := book.BuyOrders[i] if buyOrder.Price < order.Price { break } } if i == n-1 { book.BuyOrders = append(book.BuyOrders, order) } else { copy(book.BuyOrders[i+1:], book.BuyOrders[i:]) book.BuyOrders[i] = order }}// Add a sell order to the order bookfunc (book *OrderBook) addSellOrder(order Order) { n := len(book.SellOrders) var i int for i := n - 1; i >= 0; i-- { sellOrder := book.SellOrders[i] if sellOrder.Price > order.Price { break } } if i == n-1 { book.SellOrders = append(book.SellOrders, order) } else { copy(book.SellOrders[i+1:], book.SellOrders[i:]) book.SellOrders[i] = order }}// Remove a buy order from the order book at a given indexfunc (book *OrderBook) removeBuyOrder(index int) { book.BuyOrders = append(book.BuyOrders[:index], book.BuyOrders[index+1:]...)}// Remove a sell order from the order book at a given indexfunc (book *OrderBook) removeSellOrder(index int) { book.SellOrders = append(book.SellOrders[:index], book.SellOrders[index+1:]...)}
package engine// Process an order and return the trades generated before adding the remaining amount to the marketfunc (book *OrderBook) Process(order Order) []Trade { if order.Side == 1 { return book.processLimitBuy(order) } return book.processLimitSell(order)}// Process a limit buy orderfunc (book *OrderBook) processLimitBuy(order Order) []Trade { trades := make([]Trade, 0, 1) n := len(book.SellOrders) // check if we have at least one matching order if n != 0 || book.SellOrders[n-1].Price <= order.Price { // traverse all orders that match for i := n - 1; i >= 0; i-- { sellOrder := book.SellOrders[i] if sellOrder.Price > order.Price { break } // fill the entire order if sellOrder.Amount >= order.Amount { trades = append(trades, Trade{order.ID, sellOrder.ID, order.Amount, sellOrder.Price}) sellOrder.Amount -= order.Amount if sellOrder.Amount == 0 { book.removeSellOrder(i) } return trades } // fill a partial order and continue if sellOrder.Amount < order.Amount { trades = append(trades, Trade{order.ID, sellOrder.ID, sellOrder.Amount, sellOrder.Price}) order.Amount -= sellOrder.Amount book.removeSellOrder(i) continue } } } // finally add the remaining order to the list book.addBuyOrder(order) return trades}// Process a limit sell orderfunc (book *OrderBook) processLimitSell(order Order) []Trade { trades := make([]Trade, 0, 1) n := len(book.BuyOrders) // check if we have at least one matching order if n != 0 || book.BuyOrders[n-1].Price >= order.Price { // traverse all orders that match for i := n - 1; i >= 0; i-- { buyOrder := book.BuyOrders[i] if buyOrder.Price < order.Price { break } // fill the entire order if buyOrder.Amount >= order.Amount { trades = append(trades, Trade{order.ID, buyOrder.ID, order.Amount, buyOrder.Price}) buyOrder.Amount -= order.Amount if buyOrder.Amount == 0 { book.removeBuyOrder(i) } return trades } // fill a partial order and continue if buyOrder.Amount < order.Amount { trades = append(trades, Trade{order.ID, buyOrder.ID, buyOrder.Amount, buyOrder.Price}) order.Amount -= buyOrder.Amount book.removeBuyOrder(i) continue } } } // finally add the remaining order to the list book.addSellOrder(order) return trades}
package mainimport ( "engine/engine" "log" "github.com/Shopify/sarama" cluster "github.com/bsm/sarama-cluster")func main() { // create the consumer and listen for new order messages consumer := createConsumer() // create the producer of trade messages producer := createProducer() // create the order book book := engine.OrderBook{ BuyOrders: make([]engine.Order, 0, 100), SellOrders: make([]engine.Order, 0, 100), } // create a signal channel to know when we are done done := make(chan bool) // start processing orders go func() { for msg := range consumer.Messages() { var order engine.Order // decode the message order.FromJSON(msg.Value) // process the order trades := book.Process(order) // send trades to message queue for _, trade := range trades { rawTrade := trade.ToJSON() producer.Input() <- &sarama.ProducerMessage{ Topic: "trades", Value: sarama.ByteEncoder(rawTrade), } } // mark the message as processed consumer.MarkOffset(msg, "") } done <- true }() // wait until we are done <-done}//// Create the consumer//func createConsumer() *cluster.Consumer { // define our configuration to the cluster config := cluster.NewConfig() config.Consumer.Return.Errors = false config.Group.Return.Notifications = false config.Consumer.Offsets.Initial = sarama.OffsetOldest // create the consumer consumer, err := cluster.NewConsumer([]string{"127.0.0.1:9092"}, "myconsumer", []string{"orders"}, config) if err != nil { log.Fatal("Unable to connect consumer to kafka cluster") } go handleErrors(consumer) go handleNotifications(consumer) return consumer}func handleErrors(consumer *cluster.Consumer) { for err := range consumer.Errors() { log.Printf("Error: %s\n", err.Error()) }}func handleNotifications(consumer *cluster.Consumer) { for ntf := range consumer.Notifications() { log.Printf("Rebalanced: %+v\n", ntf) }}//// Create the producer//func createProducer() sarama.AsyncProducer { config := sarama.NewConfig() config.Producer.Return.Successes = false config.Producer.Return.Errors = true config.Producer.RequiredAcks = sarama.WaitForAll producer, err := sarama.NewAsyncProducer([]string{"127.0.0.1:9092"}, config) if err != nil { log.Fatal("Unable to connect producer to kafka server") } return producer}
正如你上面看到的，我们的系统将包含引擎的多个客户端，这些客户端可以是交易所 系统中的其他组件，例如接收终端用户委托请求的App等等。
在客户端和引擎之间的通信是使用Apache Kafka作为消息总线来实现的，每个交易对 都对应Kafka的一个主题，这样我们可以确保当消息队列接收到用户委托单时，引擎 将以同样的先后顺序处理委托单。这保证了即使引擎崩溃重启我们也可以重建交易 委托账本。
引擎将监听Kafka主题，执行委托账本命令并将引擎的输出事件发布到消息队列中。 当然如果能够监测委托单的处理速度以及交易的执行情况会更酷。我们可以使用 Prometheus来采集性能指标，使用grafana来实现一个监视仪表盘。