【问题标题】:How to reply an accepted connection in nats.Subscribe in golang如何在 nats 中回复已接受的连接。在 golang 中订阅
【发布时间】:2018-10-03 07:21:04
【问题描述】:

我正在尝试在 golang 中创建一个 tcp 服务器,它接受连接,读取数据,然后通过 nats.Publish 发布它。但我也想回复并关闭订阅者内部的连接,而不是在主服务器脚本中 我看到两种可能的方式:发布连接或在订阅者中建立新的连接,但我无法实现 提前谢谢你

UPD 关于 I. Kozlovic 回答的评论: 不完全是我想要的,但可能会有所帮助:) 我正在写关于关闭 tcp 连接的文章——不是 nats 连接,而是在另一个 go 进程中关闭——而不是在它被接受的那个中。但是由于我没有成功,我可以根据您的回答执行以下操作:我可以通过 nats 发布我需要的信息,对其进行处理,然后回复“OK。关闭连接”并关闭发布者中的 TCP 连接,这是一个 TCP服务器。但是,如果我不能从这个 go 过程而是从另一个过程中回复,那就太好了。我理解这很棘手,但我想做与我在上一条消息中写的相同的操作,但需要进行一些修改。我有 TCP 服务器,它接受连接,然后连接 nats 并发布消息,订阅此消息的另一个 go 进程处理输入,然后将其输出发布到处理其输入的第三个 go 进程,然后回复“OK。完成。关闭 TCP 连接" 到第一个 go 过程.... :)

【问题讨论】:

    标签: go tcpserver nats.io


    【解决方案1】:

    根据您更新的问题,这是一种可能的方法。 请注意,这两个额外的进程在这里由 go-routines 表示,但在实际情况下,您可以让它们成为单独的进程。 我也省略了错误检查。

    // This represent what would be the last process in your
    // example.
    go func() {
        nc, _ := nats.Connect(nats.DefaultURL)
        nc.Subscribe("bar", func(m *nats.Msg) {
            fmt.Printf("Received request: %s, final stop, sending back to %v\n", m.Data, m.Reply)
            nc.Publish(m.Reply, []byte("I'm here to help!"))
        })
        nc.Flush()
        runtime.Goexit()
    }()
    
    // This would be the in-between process that receives
    // the message triggered by the TCP accept
    go func() {
        nc, _ := nats.Connect(nats.DefaultURL)
        nc.Subscribe("foo", func(m *nats.Msg) {
            fmt.Printf("Received request: %s, forward to bar\n", m.Data)
            nc.PublishRequest("bar", m.Reply, []byte(fmt.Sprintf("got %s", m.Data)))
        })
        nc.Flush()
        runtime.Goexit()
    }()
    
    // This would be your TCP server
    l, _ := net.Listen("tcp", "127.0.0.1:1234")
    for {
        c, _ := l.Accept()
        go func(c net.Conn) {
            // Close socket when done
            defer c.Close()
            // Connect to NATS
            nc, _ := nats.Connect(nats.DefaultURL)
            // Close NATS connection when done
            defer nc.Close()
            // Sends the request to first process. Note that this
            // has a timeout and so if no response is received, the
            // go-routine will exit, closing the TCP connection.
            reply, err := nc.Request("foo", []byte("help"), 10*time.Second)
            if err != nil {
                fmt.Printf("Got error: %v\n", err)
            } else {
                fmt.Printf("Got reply: %s\n", reply.Data)
            }
        }(c)
    }
    

    请注意,通常不建议创建寿命很短的 NATS 连接。如果适合您的模型,您可能希望重用 NATS 连接。

    【讨论】:

    • 感谢您的回答。我通过问题更新回复了你。我的消息似乎太长,无法发表评论
    • 我明白了。在提出另一种方法之前,为什么需要保持原来的 TCP 连接打开呢?据我了解,接受的 TCP 连接只是触发 NATS 连接和发布,对吗?
    • 无论如何我都编辑了我的答案。将根据您的回答再次更新:-)
    • 太棒了!感谢您的帮助
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-03-21
    • 1970-01-01
    • 2022-08-20
    • 1970-01-01
    • 2021-07-28
    • 2020-05-06
    相关资源
    最近更新 更多