diff --git a/cmd/exchange/main.go b/cmd/exchange/main.go index 62e4e19..de8a493 100644 --- a/cmd/exchange/main.go +++ b/cmd/exchange/main.go @@ -10,6 +10,7 @@ import ( "sig-pub/pkg/config" "sig-pub/pkg/grpc/discovery" "sig-pub/pkg/grpc/interceptor" + "sig-pub/pkg/mq" "sig-pub/pkg/storage/kvrocks" vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" "sig-pub/pkg/utils/exit" @@ -33,7 +34,7 @@ func main() { conf := config.MustLoadConfig(new(config.Configuration), "config/config.toml") exchangeConf := config.MustLoadConfig(new(ExchangeConf), "config/exchange.toml") - // consul 配置 + // consul cc := api.DefaultConfig() cc.Address = conf.Consul.Address consulClient, err := api.NewClient(cc) @@ -43,6 +44,17 @@ func main() { dis := discovery.NewConsulDiscovery(consulClient) resolver := dis.Resolver() + // kvrocks db + kvdb := kvrocks.NewKVRocksDB(conf.Database.Kvrocks) + if err := kvdb.Ping(); err != nil { + panic(err) + } + + // nats + if err = mq.InitNats(conf.Nats.Url); err != nil { + panic(err) + } + // new market grpc client marketUrl := discovery.ConsulDialUrl(pb.MarketService_ServiceDesc.ServiceName) marketConn, err := grpc.NewClient(marketUrl, @@ -55,12 +67,6 @@ func main() { marketClient := pb.NewMarketServiceClient(marketConn) tradeInstanceAside := client.NewTradeInstanceAside(marketClient) - // kvrocks db - kvdb := kvrocks.NewKVRocksDB(conf.Database.Kvrocks) - if err := kvdb.Ping(); err != nil { - panic(err) - } - // okx exchange okxSubscriber := okx.NewOkxSubscriber(exchangeConf.Okx) if err := okxSubscriber.Init(); err != nil { diff --git a/cmd/trading/main.go b/cmd/trading/main.go index ff7c94a..d5a2a2b 100644 --- a/cmd/trading/main.go +++ b/cmd/trading/main.go @@ -9,6 +9,7 @@ import ( "sig-pub/pkg/config" "sig-pub/pkg/grpc/discovery" "sig-pub/pkg/grpc/interceptor" + "sig-pub/pkg/mq" "sig-pub/pkg/utils/exit" "sig-pub/pkg/zlog" @@ -29,7 +30,7 @@ func main() { conf := config.MustLoadConfig(new(config.Configuration), "config/config.toml") tradingConf := config.MustLoadConfig(new(TradingConf), "config/trading.toml") - // consul 配置 + // consul cc := api.DefaultConfig() cc.Address = conf.Consul.Address consulClient, err := api.NewClient(cc) @@ -39,6 +40,11 @@ func main() { dis := discovery.NewConsulDiscovery(consulClient) resolver := dis.Resolver() + // nats + if err = mq.InitNats(conf.Nats.Url); err != nil { + panic(err) + } + // new market grpc client marketClient, err := client.NewMarketClient( grpc.WithResolvers(resolver), diff --git a/config/config.toml b/config/config.toml index 0ab286b..081c698 100644 --- a/config/config.toml +++ b/config/config.toml @@ -20,6 +20,9 @@ password = "" [consul] address = "127.0.0.1:8500" +[nats] +url = "nats://@127.0.0.1:7222" + [database.mysql] logMode = "info" # https://gorm.io/zh_CN/docs/connecting_to_the_database.html diff --git a/docker-compose.yml b/docker-compose.yml index a9e2313..085638c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -126,3 +126,13 @@ services: - "./fs/consul/config:/consul/config" # ports: # - 8500:8500 + + sig-nats: + container_name: sig-nats + image: nats:2.12-alpine + command: -js -p 7222 -m 7221 # --auth 123456 + ports: + - "7221" + - "7222:7222" + volumes: + - ./fs/nats:/tmp/nats/jetstream diff --git a/go.mod b/go.mod index c1666f2..50cac87 100644 --- a/go.mod +++ b/go.mod @@ -12,14 +12,16 @@ require ( github.com/fanjindong/go-cache v0.0.6 github.com/gin-gonic/gin v1.10.0 github.com/go-resty/resty/v2 v2.16.5 + github.com/gogo/protobuf v1.3.2 github.com/gorilla/websocket v1.5.3 github.com/govalues/decimal v0.1.36 github.com/hashicorp/consul/api v1.32.1 github.com/influxdata/influxdb-client-go/v2 v2.14.0 github.com/jhump/protoreflect/v2 v2.0.0-beta.2 - github.com/klauspost/compress v1.17.9 + github.com/klauspost/compress v1.18.0 github.com/lib/pq v1.10.9 github.com/mostynb/go-grpc-compression v1.2.3 + github.com/nats-io/nats.go v1.47.0 github.com/redis/go-redis/v9 v9.7.3 github.com/spf13/cast v1.10.0 github.com/spf13/viper v1.20.0 @@ -27,7 +29,7 @@ require ( go.etcd.io/etcd/client/v3 v3.6.1 go.uber.org/zap v1.27.0 golang.org/x/net v0.38.0 - golang.org/x/sync v0.12.0 + golang.org/x/sync v0.13.0 golang.org/x/time v0.8.0 gonum.org/v1/gonum v0.16.0 google.golang.org/grpc v1.71.1 @@ -57,7 +59,6 @@ require ( github.com/go-sql-driver/mysql v1.7.0 // indirect github.com/go-viper/mapstructure/v2 v2.2.1 // indirect github.com/goccy/go-json v0.10.5 // indirect - github.com/gogo/protobuf v1.3.2 // indirect github.com/golang/protobuf v1.5.4 // indirect github.com/golang/snappy v0.0.4 // indirect github.com/google/uuid v1.6.0 // indirect @@ -86,6 +87,8 @@ require ( github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/nats-io/nkeys v0.4.11 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/oapi-codegen/runtime v1.0.0 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect github.com/sagikazarmark/locafero v0.7.0 // indirect @@ -100,10 +103,10 @@ require ( go.etcd.io/etcd/client/pkg/v3 v3.6.1 // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/arch v0.15.0 // indirect - golang.org/x/crypto v0.36.0 // indirect + golang.org/x/crypto v0.37.0 // indirect golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect - golang.org/x/sys v0.31.0 // indirect - golang.org/x/text v0.23.0 // indirect + golang.org/x/sys v0.32.0 // indirect + golang.org/x/text v0.24.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index c57d01c..d5f24a8 100644 --- a/go.sum +++ b/go.sum @@ -192,8 +192,8 @@ github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPci github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= -github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= @@ -246,6 +246,12 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY github.com/mostynb/go-grpc-compression v1.2.3 h1:42/BKWMy0KEJGSdWvzqIyOZ95YcR9mLPqKctH7Uo//I= github.com/mostynb/go-grpc-compression v1.2.3/go.mod h1:AghIxF3P57umzqM9yz795+y1Vjs47Km/Y2FE6ouQ7Lg= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/nats-io/nats.go v1.47.0 h1:YQdADw6J/UfGUd2Oy6tn4Hq6YHxCaJrVKayxxFqYrgM= +github.com/nats-io/nats.go v1.47.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g= +github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0= +github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/oapi-codegen/runtime v1.0.0 h1:P4rqFX5fMFWqRzY9M/3YF9+aPSPPB06IzP2P7oOxrWo= github.com/oapi-codegen/runtime v1.0.0/go.mod h1:LmCUMQuPB4M/nLXilQXhHw+BLZdDb18B34OO356yJ/A= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= @@ -356,8 +362,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34= -golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc= +golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= +golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 h1:nDVHiLt8aIbd/VzvPWN6kSOPE7+F/fNFDSXLVYkE/Iw= golang.org/x/exp v0.0.0-20250305212735-054e65f0b394/go.mod h1:sIifuuw/Yco/y6yb6+bDNfyeQ/MdPUy/hKEMYQV17cM= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= @@ -379,8 +385,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw= -golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610= +golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -404,15 +410,15 @@ golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik= -golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20= +golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= -golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= +golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0= +golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU= golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg= golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/internal/exchange/exchange.go b/internal/exchange/exchange.go index 7cc6a9c..03a128a 100644 --- a/internal/exchange/exchange.go +++ b/internal/exchange/exchange.go @@ -48,6 +48,7 @@ type ExchangeTradeInstance struct { Inst *types.TradeInstance Status atomic.Int32 // 交易产品状态, 0.初始化中 1.正常, todo status change event publish LiveKline *types.IntervalState[types.Kline] // 实时k线数据 + LastKline *types.IntervalState[types.Kline] // 最后确认k线数据 LiveKStartTs *types.IntervalState[int64] // ws开始订阅k线标记时间戳 HistoryMarkTs *types.IntervalState[int64] // 拉取历史k线标记时间戳 Last decimal.Decimal // 交易产品实时价格tick更新 diff --git a/internal/exchange/exchange_data_persist.go b/internal/exchange/exchange_data_persist.go index 4d717cd..6fc0b52 100644 --- a/internal/exchange/exchange_data_persist.go +++ b/internal/exchange/exchange_data_persist.go @@ -7,6 +7,8 @@ import ( "sig-pub/pkg/storage/kvrocks" vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" "sig-pub/pkg/types" + "sig-pub/pkg/utils/exit" + "sig-pub/pkg/zlog" ) type ExchangeDataPersist struct { @@ -26,6 +28,13 @@ func NewExchangeDataService( } func (p *ExchangeDataPersist) Init() (err error) { + // victoriametrics 强制刷盘 + exit.AddHook(func() { + zlog.Infof("force flush vmtsdb") + if err := p.vmtsdb.ForceFlush(); err != nil { + zlog.Infof("force flush vmtsdb error: ", err) + } + }, exit.WithOrderFront()) return } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index c2da483..c522588 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -8,6 +8,7 @@ import ( "sig-pub/api/pb" "sig-pub/pkg/client" "sig-pub/pkg/data" + "sig-pub/pkg/mq" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/conver" @@ -92,6 +93,7 @@ func (svc *ExchangeService) subscribeExchanges() { exchangeInst := &ExchangeTradeInstance{ Inst: tradeInst, LiveKline: types.NewIntervalState[types.Kline](), + LastKline: types.NewIntervalState[types.Kline](), LiveKStartTs: types.NewIntervalState[int64](), HistoryMarkTs: types.NewIntervalState[int64](), } @@ -152,8 +154,12 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch // 标记交易产品开始订阅k线时间 exchangeInst.LiveKStartTs.SetIf(firstKline.Interval, firstKline.Ts, func(old int64) bool { return old == 0 }) - // 标记实时k线 + // 记录实时k线 exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline) + // 记录最后确认k线 + if lastKline.Confirm { + exchangeInst.LastKline.Set(lastKline.Interval, *lastKline) + } // 记录实时价格 exchangeInst.Last = lastKline.Close @@ -244,6 +250,15 @@ func (svc *ExchangeService) initialKlines(exchange *Exchange, insts []types.Trad failed = append(failed, inst) } else { success = append(success, inst) + + // 发布交易产品初始化完成事件 + publish := &mq.PublishExchangeTradeInstanceInited{ + Exchange: inst.Exchange, + InstId: inst.InstId, + } + if err := mq.NatsPublish(mq.TopicExchangeTradeInstanceInited, publish); err != nil { + zlog.Errorf("publish trade instance inited error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) + } } } @@ -568,9 +583,14 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory if len(klines) < int(count) { rsp.Next = false } - lastK := klines[len(klines)-1] + // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 + if lastConfirmK := exchangeInst.LastKline.Get(interval); lastConfirmK.Ts == lastK.Ts { + klines[len(klines)-1] = &lastConfirmK + lastK = &lastConfirmK + } + // 降序排序 if !req.Asc { collect.Reverse(klines) diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index efb5d4c..1536716 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -5,9 +5,10 @@ import ( "io" "sig-pub/api/pb" "sig-pub/pkg/client" + "sig-pub/pkg/mq" "sig-pub/pkg/types" + "sig-pub/pkg/utils/collect" "sig-pub/pkg/zlog" - "sync" "time" "google.golang.org/grpc" @@ -19,18 +20,23 @@ type TradingService struct { klineStore *KlineStore - subKlineLock sync.Mutex - subKlineInsts [3][]string - subKlineStream grpc.BidiStreamingClient[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline] + subscribeKlineIntervals []string + subKlineInsts [3][]string + subKlineStream grpc.BidiStreamingClient[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline] } func NewTradingService( marketClientAside *client.TradeInstanceAside, exchangeClient pb.ExchangeServiceClient, ) *TradingService { + // 订阅实时k线周期列表 + subscribeKlineIntervals := collect.Map2Slice(types.SupportedIntervals, func(interval types.Interval, _ types.IntervalAdder) string { + return string(interval) + }) return &TradingService{ - marketClientAside: marketClientAside, - exchangeClient: exchangeClient, + marketClientAside: marketClientAside, + exchangeClient: exchangeClient, + subscribeKlineIntervals: subscribeKlineIntervals, } } @@ -47,15 +53,32 @@ func (svr *TradingService) Init() (err error) { svr.subKlineInsts[exInst.Exchange] = append(svr.subKlineInsts[exInst.Exchange], exInst.InstId) } + // 拉取已初始化完成交易产品, 初始化k线, 开始订阅k线 + + // 订阅交易产品初始化完成事件 + mq.NatsCreateConsumer("trading", mq.StreamExchange, mq.TopicExchangeTradeInstanceInited, func() *mq.PublishExchangeTradeInstanceInited { return new(mq.PublishExchangeTradeInstanceInited) }, + func(msg *mq.PublishExchangeTradeInstanceInited) (err error) { + // 初始化k线, 开始订阅k线 + zlog.Infof("trade instance inited: %#v", msg) + return + }) + // 订阅k线 - go svr.subscribeKlines(false) + go svr.subscribeStreamKlines(false) return } -func (svr *TradingService) subscribeKlines(reconnect bool) { +func (svr *TradingService) subscribeKlines() { + +} + +func (svr *TradingService) subscribeStreamKlines(reconnect bool) { defer func() { - svr.subKlineStream = nil - go svr.subscribeKlines(true) + if svr.subKlineStream != nil { + svr.subKlineStream.CloseSend() + svr.subKlineStream = nil + } + go svr.subscribeStreamKlines(true) }() if reconnect { @@ -63,29 +86,25 @@ func (svr *TradingService) subscribeKlines(reconnect bool) { time.Sleep(5 * time.Second) } - // svr.subKlineLock.Lock() - // defer svr.subKlineLock.Unlock() - stream, err := svr.exchangeClient.SubscribeKline(context.Background()) if err != nil { zlog.Error("subscribeKlines reqeust error: ", err) return } - - if svr.subKlineStream != nil { - svr.subKlineStream.CloseSend() - } + svr.subKlineStream = stream // 发送订阅消息 exchanges := []pb.ExchangeType{pb.ExchangeType_OKX} for _, exchange := range exchanges { instIds := svr.subKlineInsts[exchange] - + if len(instIds) == 0 { + continue + } msg := &pb.ReqStreamSubscribeKline{ SubType: pb.SubscribeType_Subscribe, Exchanges: []pb.ExchangeType{exchange}, InstIds: instIds, - Intervals: SubscribeKlineIntervals, + Intervals: svr.subscribeKlineIntervals, OnlyConfirm: true, } zlog.Debugf("send stream subscribe kline msg: %#v", msg) @@ -116,23 +135,3 @@ func (svr *TradingService) subscribeKlines(reconnect bool) { } } } - -var SubscribeKlineIntervals = []string{ - // string(types.Interval1s), - string(types.Interval1m), - string(types.Interval3m), - string(types.Interval5m), - string(types.Interval15m), - string(types.Interval30m), - string(types.Interval1h), - string(types.Interval2h), - string(types.Interval4h), - string(types.Interval6h), - string(types.Interval12h), - string(types.Interval1d), - string(types.Interval1d), - string(types.Interval2d), - string(types.Interval3d), - string(types.Interval5d), - string(types.Interval1w), -} diff --git a/pkg/config/config.go b/pkg/config/config.go index 03b82bd..5f5788b 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -19,6 +19,7 @@ type Configuration struct { // Consul api.Config Consul ConsulConfig Database Database + Nats NatsConfig Tsdb TsdbConfig } @@ -50,6 +51,10 @@ type GateConf struct { WsAddr string } +type NatsConfig struct { + Url string +} + // ============== database ============== type Database struct { Mysql MysqlConfig diff --git a/pkg/mq/nats.go b/pkg/mq/nats.go new file mode 100644 index 0000000..5e73da2 --- /dev/null +++ b/pkg/mq/nats.go @@ -0,0 +1,383 @@ +package mq + +import ( + "context" + "errors" + "fmt" + "runtime/debug" + "sig-pub/pkg/zlog" + "strings" + "time" + + "github.com/bytedance/sonic" + "github.com/gogo/protobuf/proto" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +var ( + NatsConn *nats.Conn + NatsJS jetstream.JetStream +) + +func InitNats(natsUrl string) (err error) { + options := nats.Options{ + Url: natsUrl, + // 断线重连配置 + RetryOnFailedConnect: true, + AllowReconnect: true, + MaxReconnect: -1, + ReconnectWait: time.Second * 2, + ReconnectBufSize: 10 * 1024 * 1024, // 10M + } + + NatsConn, err = options.Connect() + if err != nil { + return + } + NatsJS, err = jetstream.New(NatsConn) + if err != nil { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() + err = initNatsStream(ctx) + if err != nil { + return + } + return +} + +// initNatsStream 初始化 jetstream 消息流 +func initNatsStream(ctx context.Context) (err error) { + for _, ss := range initialStream { + if len(ss) < 2 { + err = fmt.Errorf("initial jetstream %v config error", ss) + return + } + err = NatsCreateStream(ctx, ss[0], ss[1:]...) + if err != nil { + return + } + } + return +} + +// NatsCreateStream 创建或更新jetstream消息流 +func NatsCreateStream(ctx context.Context, streamName string, subjects ...string) (err error) { + // 活动消息流 https://docs.nats.io/nats-concepts/jetstream/streams#configuration + streamConfig := jetstream.StreamConfig{ + Name: streamName, + Storage: jetstream.FileStorage, + Subjects: subjects, // 要绑定的topics + Replicas: 1, // 消息副本数量 + MaxAge: time.Hour * 24, // 流中任何消息的最大年龄 + MaxBytes: -1, // 存储的所有消息最大字节数 -1无限制 + MaxMsgs: -1, // 流中存储的所有消息最大消息数 + MaxMsgSize: -1, // 单个消息最大字节数 + MaxConsumers: -1, // Stream 定义的最大消费者数量 + NoAck: false, + Retention: jetstream.LimitsPolicy, // 声明流的保留策略 + Discard: jetstream.DiscardOld, // stream消息数量达到limit处理策略 + MaxMsgsPerSubject: -1, // stream保留每个主题最大消息数 + Duplicates: time.Minute * 2, // 跟踪重复消息的窗口 + AllowRollup: false, // 允许使用标题Nats-Rollup将流的所有内容或流中的主题替换为单个新消息 + DenyDelete: true, // 不允许删除消息 + DenyPurge: true, // 不允许清空消息 + } + _, err = NatsJS.CreateOrUpdateStream(ctx, streamConfig) + if err != nil { + return + } + return +} + +// NatsCreateConsumer 创建或更新 jetstream consumer +// consumerName 多个相同name只消费一次,多个不同name消费分别消费一次 +func NatsCreateConsumer[T any](consumerName, streamName, topic string, inst func() T, handler func(T) (err error), options ...NatsConsumerOption) { + if NatsJS == nil { + panic(errors.New("nats js not connect")) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second*3) + defer cancel() + + consumerName = strings.ReplaceAll(fmt.Sprintf("%s-%s", consumerName, topic), ".", "-") + // https://docs.nats.io/nats-concepts/jetstream/consumers#general + consumerConfig := jetstream.ConsumerConfig{ + Durable: consumerName, + Name: consumerName, + FilterSubject: topic, + DeliverPolicy: jetstream.DeliverNewPolicy, // 只处理consumer创建后的新消息 + AckPolicy: jetstream.AckExplicitPolicy, // 消息确认:不确认高qps + ReplayPolicy: jetstream.ReplayInstantPolicy, // 消息回放机制 + MaxDeliver: -1, // 消息投递的最大次数-1表示没ack就不停投递 + MaxAckPending: -1, // 多少条消息未投递成功则不再投递新消息-1不限制 + } + for _, opt := range options { + opt(&consumerConfig) + } + consumer, err := NatsJS.Consumer(ctx, streamName, consumerName) + if err != nil { + consumer, err = NatsJS.CreateConsumer(ctx, streamName, consumerConfig) + if err != nil { + // consumer 配置修改 + if err == jetstream.ErrConsumerExists { + consumer, err = NatsJS.UpdateConsumer(ctx, streamName, consumerConfig) + if err != nil { + // NatsJS.DeleteConsumer(ctx, streamName, consumerName) + panic(err) + } + } else { + panic(err) + } + } + } + consumer.Consume(func(msg jetstream.Msg) { + defer func() { + if r := recover(); r != nil { + zlog.Errorf("nats consumer recover error: consumer=%s, stream=%s, topic=%s, %v", consumerName, streamName, topic, r) + debug.PrintStack() + } + }() + + defer msg.Ack() + + var data T + if inst != nil { + data = inst() + if err := natsJsUnmarshal(data, msg); err != nil { + zlog.Errorf("unmarshal data %T error, %s, %v", data, msg.Subject(), err) + return + } + } + err = handler(data) + if err != nil { + zlog.Error(err) + debug.PrintStack() + } + }) +} + +func NatsCreateConsumerExtend[T any](consumerName, streamName, topic string, inst func() T, handler func(T, *jetstream.MsgMetadata) (ack bool, delay time.Duration, err error), options ...NatsConsumerOption) { + if NatsJS == nil { + panic(errors.New("nats js not connect")) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second*3) + defer cancel() + + consumerName = strings.ReplaceAll(fmt.Sprintf("%s-%s", consumerName, topic), ".", "-") + // https://docs.nats.io/nats-concepts/jetstream/consumers#general + consumerConfig := jetstream.ConsumerConfig{ + Durable: consumerName, + Name: consumerName, + FilterSubject: topic, + DeliverPolicy: jetstream.DeliverNewPolicy, // 只处理consumer创建后的新消息 + AckPolicy: jetstream.AckExplicitPolicy, // 消息确认:不确认高qps + ReplayPolicy: jetstream.ReplayInstantPolicy, // 消息回放机制 + MaxDeliver: -1, // 消息投递的最大次数-1表示没ack就不停投递 + MaxAckPending: -1, // 多少条消息未投递成功则不再投递新消息-1不限制 + } + for _, opt := range options { + opt(&consumerConfig) + } + consumer, err := NatsJS.Consumer(ctx, streamName, consumerName) + if err != nil { + consumer, err = NatsJS.CreateConsumer(ctx, streamName, consumerConfig) + if err != nil { + // consumer 配置修改 + if err == jetstream.ErrConsumerExists { + consumer, err = NatsJS.UpdateConsumer(ctx, streamName, consumerConfig) + if err != nil { + // NatsJS.DeleteConsumer(ctx, streamName, consumerName) + panic(err) + } + } else { + panic(err) + } + } + } + consumer.Consume(func(msg jetstream.Msg) { + defer func() { + if r := recover(); r != nil { + zlog.Errorf("nats consumer recover error: consumer=%s, stream=%s, topic=%s, %v", consumerName, streamName, topic, r) + debug.PrintStack() + } + }() + + var data T + if inst != nil { + data = inst() + if err := natsJsUnmarshal(data, msg); err != nil { + zlog.Errorf("unmarshal data %T error, %s, %v", data, msg.Subject(), err) + msg.Ack() // 解析失败直接确认，避免无限重试 + return + } + } + + // 获取消息元数据 + metadata, err := msg.Metadata() + if err != nil { + zlog.Errorf("get message metadata error: %v", err) + msg.Ack() // 获取元数据失败直接确认 + return + } + + // 调用handler并处理结果 + ack, delay, err := handler(data, metadata) + if err != nil { + zlog.Error(err) + debug.PrintStack() + } + + // 根据返回值决定确认或重试 + if ack { + msg.Ack() + } else { + if err := msg.NakWithDelay(delay); err != nil { + zlog.Errorf("nak with delay error: %v", err) + msg.Nak() // 如果设置延迟失败，直接Nak + } + } + }) +} + +// NatsSubscribe nats 请求响应注册 +func NatsSubscribe[T, V any](subject string, inst func() T, handler func(req T) (rsp V, err error)) (s *nats.Subscription) { + if NatsConn == nil { + panic(errors.New("nats not connect")) + } + s, err := NatsConn.Subscribe(subject, func(msg *nats.Msg) { + defer func() { + if r := recover(); r != nil { + zlog.Errorf("nats subject recover error: topic=%s, %v", subject, r) + debug.PrintStack() + } + }() + defer msg.Ack() + + var req T + if inst != nil { + req = inst() + if err := natsUnmarshal(req, msg); err != nil { + zlog.Errorf("unmarshal data %T error, %s", req, msg.Subject) + return + } + } + + rsp, err := handler(req) + + rspMsg := nats.NewMsg(subject) + if err != nil { + rspMsg.Header.Add("error", err.Error()) + } else { + if err := natsMarshal(rsp, rspMsg); err != nil { + zlog.Errorf("marshal rsp data %T error, %s", rsp, msg.Subject) + return + } + } + msg.RespondMsg(rspMsg) + }) + if err != nil { + panic(err) + } + return +} + +// NatsRequest nats发送请求 +func NatsRequest(subject string, rsp, req any, options ...NatsRequestOption) (err error) { + if NatsConn == nil { + return errors.New("nats not connect") + } + c := &NatsRequestOptions{Timeout: time.Second * 10} + for _, option := range options { + option(c) + } + reqMsg := nats.NewMsg(subject) + if err = natsMarshal(req, reqMsg); err != nil { + return + } + rspMsg, err := NatsConn.RequestMsg(reqMsg, c.Timeout) + if err != nil { + return + } + if e := rspMsg.Header.Get("error"); e != "" { + err = errors.New(e) + return + } + if err = natsUnmarshal(rsp, rspMsg); err != nil { + return + } + return +} + +// NatsPublish nats 发布消息 +func NatsPublish(subject string, data any) (err error) { + if NatsJS == nil { + return errors.New("nats js not connect") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + + msg := nats.NewMsg(subject) + if err = natsMarshal(data, msg); err != nil { + return + } + _, err = NatsJS.PublishMsg(ctx, msg) + if err != nil { + return + } + return +} + +func natsMarshal(data any, msg *nats.Msg) (err error) { + mt := "1" // 序列化1json,2protobuf + if marshaler, ok := data.(proto.Marshaler); ok { + if bytes, err := marshaler.Marshal(); err == nil { + mt = "2" + msg.Data = bytes + } + } + if mt == "1" { + msg.Data, err = sonic.Marshal(data) + if err != nil { + return + } + } + msg.Header.Add("mt", mt) + return +} + +func natsUnmarshal(data any, msg *nats.Msg) (err error) { + mt := msg.Header.Get("mt") + if mt == "2" { + unmarshaler, ok := data.(proto.Unmarshaler) + if !ok { + return fmt.Errorf("%T is not protobuf type", data) + } + return unmarshaler.Unmarshal(msg.Data) + } + // mt=1 json + err = sonic.Unmarshal(msg.Data, data) + if err != nil { + return + } + return +} + +func natsJsUnmarshal(data any, msg jetstream.Msg) (err error) { + mt := msg.Headers().Get("mt") + if mt == "2" { + unmarshaler, ok := data.(proto.Unmarshaler) + if !ok { + return fmt.Errorf("%T is not protobuf type", data) + } + return unmarshaler.Unmarshal(msg.Data()) + } + // mt=1 json + err = sonic.Unmarshal(msg.Data(), data) + if err != nil { + return + } + return +} diff --git a/pkg/mq/nats_data.go b/pkg/mq/nats_data.go new file mode 100644 index 0000000..c1d91f8 --- /dev/null +++ b/pkg/mq/nats_data.go @@ -0,0 +1,9 @@ +package mq + +import "sig-pub/api/pb" + +// 交易产品初始化完成通知 +type PublishExchangeTradeInstanceInited struct { + Exchange pb.ExchangeType `json:"exchange"` // 交易所类型 + InstId string `json:"instId"` // 交易产品id +} diff --git a/pkg/mq/nats_options.go b/pkg/mq/nats_options.go new file mode 100644 index 0000000..6c90a6d --- /dev/null +++ b/pkg/mq/nats_options.go @@ -0,0 +1,40 @@ +package mq + +import ( + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +type NatsConsumerOption func(c *jetstream.ConsumerConfig) + +type NatsRequestOption func(c *NatsRequestOptions) + +type NatsRequestOptions struct { + Timeout time.Duration +} + +// WithConsumerNoneAck jetstream 消息不ack高qps +func WithConsumerNoneAck() NatsConsumerOption { + return func(c *jetstream.ConsumerConfig) { + c.AckPolicy = jetstream.AckNonePolicy + } +} + +func WithConsumerMaxDeliver(maxDeliver int) NatsConsumerOption { + return func(c *jetstream.ConsumerConfig) { + c.MaxDeliver = maxDeliver + } +} + +func WithConsumerMaxAckPending(maxAckPending int) NatsConsumerOption { + return func(c *jetstream.ConsumerConfig) { + c.MaxAckPending = maxAckPending + } +} + +func WithRequestTimeout(timeout time.Duration) NatsRequestOption { + return func(c *NatsRequestOptions) { + c.Timeout = timeout + } +} diff --git a/pkg/mq/nats_topic.go b/pkg/mq/nats_topic.go new file mode 100644 index 0000000..bd86b00 --- /dev/null +++ b/pkg/mq/nats_topic.go @@ -0,0 +1,22 @@ +package mq + +// 初始化jetStream流, {streamName, topics...} +var initialStream = [][]string{ + {StreamExchange, "exchange.>"}, +} + +const ( + // ========================== jetstraem定义 ========================== + StreamExchange = "stream-exchange" // 游戏行为jetstraem消息流 + + // ========================== mq主题订阅 ========================== + // ------------- 交易所 ------------- + TopicExchangeTradeInstanceInited = "exchange.trade_instance_inited" // 交易产品初始化完成 + + // ========================== 请求响应 ========================== + // ------------- 拉去历史k线 ------------- + // RequestExchangeHistoryKline = "request.exchange.history_kline" +) + +// nats 空参数 +var NatsNilData = map[string]any{} diff --git a/pkg/utils/collect/collect.go b/pkg/utils/collect/collect.go index 95db500..69cf3a7 100644 --- a/pkg/utils/collect/collect.go +++ b/pkg/utils/collect/collect.go @@ -69,15 +69,37 @@ func Tail[T any](slice []T, dv T) T { return slice[len(slice)-1] } -func MapValues[K comparable, V any](kvs map[K]V) []V { - var values []V +func MapKeys[K comparable, V any](kvs map[K]V) (keys []K) { + if len(kvs) == 0 { + return + } + keys = make([]K, 0, len(kvs)) + for k := range kvs { + keys = append(keys, k) + } + return +} + +func MapValues[K comparable, V any](kvs map[K]V) (values []V) { if len(kvs) == 0 { return values } + values = make([]V, 0, len(kvs)) for _, v := range kvs { values = append(values, v) } - return values + return +} + +func Map2Slice[K comparable, V any, T any](kvs map[K]V, fn func(k K, v V) T) (values []T) { + if len(kvs) == 0 { + return values + } + values = make([]T, 0, len(kvs)) + for k, v := range kvs { + values = append(values, fn(k, v)) + } + return } func Sum[T int32 | int64 | int](nums []T) T { @@ -153,17 +175,6 @@ func SliceEquals[T comparable](slice1 []T, slice2 []T) bool { return true } -func Join(slice []string, on string) (res string) { - length := len(slice) - for i := 0; i < length; i++ { - res += slice[i] - if i < length-1 { - res += on - } - } - return -} - func CopyMap[K comparable, V any](src map[K]V) (dst map[K]V) { dst = make(map[K]V, len(src)) for k, v := range src { @@ -172,19 +183,7 @@ func CopyMap[K comparable, V any](src map[K]V) (dst map[K]V) { return } -func CopyMap2[K comparable, K2 comparable, V2 any](src map[K]map[K2]V2) (dst map[K]map[K2]V2) { - dst = make(map[K]map[K2]V2, len(src)) - for k, v := range src { - cv := make(map[K2]V2, len(v)) - for k2, v2 := range v { - cv[k2] = v2 - } - dst[k] = cv - } - return -} - -func IndexMapping[T any](slice []T) (index []int) { +func MappingIndex[T any](slice []T) (index []int) { index = make([]int, len(slice)) for i := range slice { index[i] = i @@ -193,23 +192,13 @@ func IndexMapping[T any](slice []T) (index []int) { } func SortIndex[T any](slice []T, less func(i, j int) bool) (sortIndex []int) { - sortIndex = IndexMapping(slice) + sortIndex = MappingIndex(slice) sort.Slice(sortIndex, func(i, j int) bool { return less(sortIndex[i], sortIndex[j]) }) return } -func SliceMapping[T interface{}, M interface{}](slice []T, mapping func(T) M) (m []M) { - if len(slice) == 0 { - return - } - for _, item := range slice { - m = append(m, mapping(item)) - } - return -} - func Reverse[T any](slice []T) { length := len(slice) for i := range length / 2 {