chenyh
/
vRp.CD2g_test
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

基于Nats的跨服实现

This commit is contained in:
kercylan98 2023-05-16 11:40:22 +08:00
parent fc9d334695
commit 6c088cf8b4
1 changed files with 2 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
server/corss/nats.go
View File

@ -34,18 +34,16 @@ type Nats struct {
conn *nats.Conn conn *nats.Conn
url string url string
subject string subject string
realSubject string
options []nats.Option options []nats.Option
messagePool *synchronization.Pool[*Message] messagePool *synchronization.Pool[*Message]
} }
func (slf *Nats) Init(server *server.Server, packetHandle func(serverId int64, packet []byte)) (err error) { func (slf *Nats) Init(server *server.Server, packetHandle func(serverId int64, packet []byte)) (err error) {
slf.realSubject = fmt.Sprintf("%s_%d", slf.subject, server.GetID())
slf.conn, err = nats.Connect(slf.url, slf.options...) slf.conn, err = nats.Connect(slf.url, slf.options...)
if err != nil { if err != nil {
return err return err
} }
_, err = slf.conn.Subscribe(slf.realSubject, func(msg *nats.Msg) { _, err = slf.conn.Subscribe(fmt.Sprintf("%s_%d", slf.subject, server.GetID()), func(msg *nats.Msg) {
message := slf.messagePool.Get() message := slf.messagePool.Get()
defer slf.messagePool.Release(message) defer slf.messagePool.Release(message)
if err := json.Unmarshal(msg.Data, &message); err != nil { if err := json.Unmarshal(msg.Data, &message); err != nil {
@ -66,5 +64,5 @@ func (slf *Nats) PushMessage(serverId int64, packet []byte) error {
if err != nil { if err != nil {
return err return err
} }
return slf.conn.Publish(slf.realSubject, data) return slf.conn.Publish(fmt.Sprintf("%s_%d", slf.subject, serverId), data)
} }