我目前正在研究微服务架构。
在将NATS插入项目之前，我想用它测试一些简单的场景。
在一种情况下，我有一个简单的发布者，该发布者通过在localhost:4222上运行的基本Nats服务器在for循环中发布100.000条消息。
最大的问题是订户。当他收到30.000-40.000条消息时，我的整个main.go程序和所有其他go例程都停止了，什么也不做。我可以使用ctrl + c退出。但是发布者仍在继续发送邮件。当我打开一个新终端并启动该订户的一个新实例时，所有工作再次正常，直到订户收到大约30000条消息。最糟糕的是，服务器上甚至没有出现一个错误，也没有日志，所以我不知道发生了什么。
之后，我尝试将Subscribe方法替换为QueueSubscribe方法，并且一切正常。
订阅和队列订阅之间的主要区别是什么？
NATS-Streaming是更好的机会吗？或者在某些情况下，我应该更喜欢流媒体，而在标准NATS服务器中
这是我的代码:
发布者:
package main
import (
"fmt"
"log"
"time"
"github.com/nats-io/go-nats"
)
func main() {
go createPublisher()
for {
}
}
func createPublisher() {
log.Println("pub started")
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
msg := make([]byte, 16)
for i := 0; i < 100000; i++ {
nc.Publish("alenSub", msg)
if (i % 100) == 0 {
fmt.Println("i", i)
}
time.Sleep(time.Millisecond)
}
log.Println("pub finish")
nc.Flush()
}
订户:
package main
import (
"fmt"
"log"
"time"
"github.com/nats-io/go-nats"
)
var received int64
func main() {
received = 0
go createSubscriber()
go check()
for {
}
}
func createSubscriber() {
log.Println("sub started")
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
nc.Subscribe("alenSub", func(msg *nats.Msg) {
received++
})
nc.Flush()
for {
}
}
func check() {
for {
fmt.Println("-----------------------")
fmt.Println("still running")
fmt.Println("received", received)
fmt.Println("-----------------------")
time.Sleep(time.Second * 2)
}
}
最佳答案
无限的for循环可能会使垃圾收集器饿死:https://github.com/golang/go/issues/15442#issuecomment-214965471
通过运行发布者，我能够重现该问题。要解决此问题，我建议使用sync.WaitGroup。这是我更新注释中链接的代码以使其完成的方式:
package main
import (
"fmt"
"log"
"sync"
"time"
"github.com/nats-io/go-nats"
)
// create wait group
var wg sync.WaitGroup
func main() {
// add 1 waiter
wg.Add(1)
go createPublisher()
// wait for wait group to complete
wg.Wait()
}
func createPublisher() {
log.Println("pub started")
// mark wait group done after createPublisher completes
defer wg.Done()
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
msg := make([]byte, 16)
for i := 0; i < 100000; i++ {
if errPub := nc.Publish("alenSub", msg); errPub != nil {
panic(errPub)
}
if (i % 100) == 0 {
fmt.Println("i", i)
}
time.Sleep(time.Millisecond * 1)
}
log.Println("pub finish")
errFlush := nc.Flush()
if errFlush != nil {
panic(errFlush)
}
errLast := nc.LastError()
if errLast != nil {
panic(errLast)
}
}
我建议类似地更新上述订户代码。
Subscribe和QueueSubscriber之间的主要区别在于，在Subscribe中，向所有订户发送所有消息。在QueueSubscribe中，每条消息仅发送一个QueueGroup中的订户。有关NATS流附加功能的一些详细信息，请点击此处:
https://nats.io/documentation/streaming/nats-streaming-intro/
我们看到从数据管道到控制平面的各种用例中都使用了NATS和NATS流。您的选择应取决于用例的需求。