向nats组提出请求

Question

下面是我的sub.go示例：

package main

import (
    "fmt"
    nats "github.com/nats-io/nats.go"
)

type Message struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

type Response struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

var nc *nats.Conn
var c *nats.EncodedConn

func start(){
    nc, _ := nats.Connect("127.0.0.1:4222")
    c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
    c.QueueSubscribe("subject_toto", "queue_titi", func(_, reply string, message  *Message) {
        fmt.Printf("%+v\n", message)
        var response Response
        response.Msg = "message received"
        response.Status = "Ok"
        c.Publish("reply", response)
        c.Flush()
    })
    c.Flush()
}

func main() {
    fmt.Println("begin")
    go start()
    defer c.Close()
    fmt.Scanln()
    nc.Drain()
    // Close connection
    nc.Close()
    fmt.Println("done")
}

它运行得很好，所以现在我想向这个队列发布一条消息，这是我的pub.go：

package main

import (
    "fmt"
    nats "github.com/nats-io/nats.go"
    "time"
)

type Message struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

type Response struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

var nc *nats.Conn
var c *nats.EncodedConn

func start(){
    var err error
    var message Message
    var response Response
    nc, _ := nats.Connect("127.0.0.1:4222")
    c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
    message.Status = "Ok"
    message.Msg = "hello"
    err = c.Request("subject_toto", message, &response, 6*time.Second)
    if err != nil {
        fmt.Printf("%+v\n", err)
    }
    fmt.Printf("%+v\n","response")
    fmt.Printf("%+v\n", response)
    defer c.Close()
}

func main() {
    fmt.Println("begin")
    start()
    fmt.Println("done")
}

但是，当我试图向它发表时，我有一个空洞的答复：

响应{状态: Msg:}

答案的开头之一似乎使用了PublishRequest，但在我看来，我只能发送字符串到服务器，而不能发送结构。

Answer 1

问题在于这一行：

c.Publish("reply", response)

当您发送请求和您的订户要作出回应时，应该有一个“回复-对主题”，即所谓的收件箱。此收件箱设置在处理程序函数的reply arg中。

因此，您必须将响应发布到reply函数的QueueSubscribe()中的处理程序函数的主题中，因此将该行更改为：

c.Publish(reply, response)

答复arg的价值对于交流很重要，它类似于：_INBOX.bw5EtJShBTI9OQdvxFOBlz.VxsGBcjH

Answer 2

以下是sub的功能版本：

package main

import (
    "encoding/json"
    "fmt"
    nats "github.com/nats-io/nats.go"
)
type Message struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

type Response struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

var nc *nats.Conn
var c *nats.EncodedConn


func start(){

    nc, _ := nats.Connect("127.0.0.1:4222")
    c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
    c.QueueSubscribe("subject_toto", "queue_titi",func(msg *nats.Msg) {
        var message Message
        err := json.Unmarshal([]byte(msg.Data), &message)
        if err != nil {
            fmt.Printf("%+v\n", err)
        }
        fmt.Printf("%+v\n", "message from pub")
        fmt.Printf("%+v\n", message)
        var response Response
        response.Msg = "message received"
        response.Status = "Ok"
        fmt.Printf("%+v\n", "response to  sub")
        fmt.Printf("%+v\n", response)
        byteConfApi, err2 := json.Marshal(response)
        if err2 != nil {
            fmt.Printf("%+v\n", err2)
        }
        msg.Respond(byteConfApi)
        c.Flush()
    })
}



func main(){
    fmt.Println("begin")
    go  start()
    defer c.Close()
    fmt.Scanln()
    nc.Drain()
    // Close connection
    nc.Close()
    fmt.Println("done")
}