Giới thiệu
Khi tôi bắt đầu xây dựng hệ thống giao dịch tần suất cao (HFT) vào năm 2024, thách thức lớn nhất không phải là thuật toán giao dịch mà là cách thu thập và xử lý dữ liệu order book từ Binance một cách ổn định, nhanh chóng và tiết kiệm chi phí. Trong bài viết này, tôi sẽ chia sẻ kiến trúc production đã xử lý hơn 50 triệu messages/ngày với độ trễ trung bình chỉ 12ms.
Bài viết này phù hợp với kỹ sư backend có kinh nghiệm muốn xây dựng hệ thống thu thập dữ liệu thị trường crypto chuyên nghiệp.
Tại Sao Cần Kiến Trúc Riêng Cho Order Book Data?
Binance WebSocket API cung cấp dữ liệu real-time qua nhiều endpoint. Tuy nhiên, khi cần thu thập đồng thời hàng chục cặp trading, raw implementation sẽ gặp các vấn đề nghiêm trọng:
- Connection limit: Mặc định giới hạn 5 connections/IP
- Message flooding: Order book update có thể đến 100+ messages/giây/cặp
- Reconnection storm: Khi connection drop, hàng ngàn reconnect đồng thời sẽ gây DDoS chính mình
- Memory leak: Không quản lý buffer đúng cách sẽ trigger OOM
Kiến Trúc Tổng Quan
+-------------------+ +-------------------+ +-------------------+
| Binance WS API |---->| Connection Pool |---->| Message Router |
| (Multiple Streams)| | (Weighted Round | | (Goroutines) |
| | | Robin + Health) | | |
+-------------------+ +-------------------+ +-------------------+
|
+-------------------+ v
| Order Book |<----+-------------------+
| Aggregator | | Backpressure |
| (LevelDB Cache) | | (Token Bucket) |
+-------------------+ +-------------------+
|
+-------------------+
| Data Processor |
| (Consumer) |
+-------------------+
|
+-------------------+
| Persistence |
| (ClickHouse/TS) |
+-------------------+
Core Implementation
1. WebSocket Manager Với Connection Pool
Đây là trái tim của hệ thống. Tôi sử dụng gorilla/websocket với custom connection pool có health check và automatic failover.
package binancews
import (
"context"
"encoding/json"
"fmt"
"log"
"sync"
"sync/atomic"
"time"
"github.com/gorilla/websocket"
)
type StreamType string
const (
StreamDepth StreamType = "@depth@100ms"
StreamDepthDiff StreamType = "@depth@100ms@100ms"
StreamTrade StreamType = "@trade"
StreamKline StreamType = "@kline_1m"
)
type Config struct {
Streams []StreamType
Symbols []string
MaxConnections int
ReconnectDelay time.Duration
HealthCheckInt time.Duration
MsgBufferSize int
WriteTimeout time.Duration
ReadDeadline time.Duration
}
type Connection struct {
ID int64
URL string
Conn *websocket.Conn
IsHealthy atomic.Bool
LastPingAt atomic.Int64
MsgReceived atomic.Int64
ReconnectCnt atomic.Int32
mu sync.RWMutex
}
type WebSocketManager struct {
config Config
connections map[string]*Connection
subscribeMsg map[string][]byte
handler MessageHandler
reconnectChan chan string
done chan struct{}
wg sync.WaitGroup
mu sync.RWMutex
// Metrics
totalMsgs atomic.Int64
reconnects atomic.Int64
errors atomic.Int64
avgLatency atomic.Int64
}
type MessageHandler func(symbol string, msgType string, data []byte, recvTime time.Time)
func NewWebSocketManager(cfg Config, handler MessageHandler) *WebSocketManager {
return &WebSocketManager{
config: cfg,
connections: make(map[string]*Connection),
subscribeMsg: make(map[string][]byte),
handler: handler,
reconnectChan: make(chan string, 1000),
done: make(chan struct{}),
}
}
func (wm *WebSocketManager) Start(ctx context.Context) error {
// Initialize connections for each stream
for _, stream := range wm.config.Streams {
for _, symbol := range wm.config.Symbols {
key := fmt.Sprintf("%s_%s", symbol, stream)
// Weighted load balancing - Binance has different DC endpoints
endpoint := wm.selectEndpoint(symbol)
url := fmt.Sprintf("%s/%s%s", endpoint, symbol, stream)
conn := &Connection{
ID: time.Now().UnixNano(),
URL: url,
}
wm.mu.Lock()
wm.connections[key] = conn
wm.mu.Unlock()
// Build subscribe message
wm.subscribeMsg[key] = wm.buildSubscribeMsg(symbol, stream)
// Start connection goroutine
wm.wg.Add(1)
go wm.runConnection(ctx, key, conn)
}
}
// Start reconnection handler
wm.wg.Add(1)
go wm.handleReconnect(ctx)
// Start metrics reporter
wm.wg.Add(1)
go wm.reportMetrics(ctx)
return nil
}
func (wm *WebSocketManager) selectEndpoint(symbol string) string {
// Binance uses different data centers for different symbol groups
// USDT-M futures: fstream.binance.com
// Spot: stream.binance.com
if isFuturesSymbol(symbol) {
return "wss://fstream.binance.com/ws"
}
return "wss://stream.binance.com:9443/ws"
}
func (wm *WebSocketManager) runConnection(ctx context.Context, key string, conn *Connection) {
defer wm.wg.Done()
for {
select {
case <-ctx.Done():
return
case <-wm.done:
return
default:
}
err := wm.connect(ctx, key, conn)
if err != nil {
wm.errors.Add(1)
wm.reconnects.Add(1)
log.Printf("[ERROR] Connection %s failed: %v, reconnecting...", key, err)
// Exponential backoff with jitter
delay := wm.config.ReconnectDelay * time.Duration(1<<