Go：将 gob 与 zmq4 结合使用

一分耕耘，一分收获！既然都打开这篇《Go：将 gob 与 zmq4 结合使用》，就坚持看下去，学下去吧！本文主要会给大家讲到等等知识点，如果大家对本文有好的建议或者看到有不足之处，非常欢迎大家积极提出！在后续文章我会继续更新Golang相关的内容，希望对大家都有所帮助！

我正在尝试将 gob 与 zmq4 套接字（来自 pebbe 的 zmq4）一起使用。 zmq4 套接字没有 io 设备，这使得 gob 似乎不可能直接读/写：

我无法在 gob.newencoder 的参数中使用 &client （类型 **zmq4.socket ） 作为 io.writer 类型：zmq4.socket 未实现 io.writer （缺少 write 方法）

一个 zmq4 发送函数 sendmessage() 接受 interface{}，因此我使用它来发送。

在服务器端，zmq4 接收函数返回 string、[]byte、[]string 或 [][]byte。我正在使用 recvmessage() 它返回 []string。

可以写入 bytes.buffer，发送该缓冲区，将其读取为 []string，然后获取消息的内容部分以使用 gob 进行处理。尽管当前的问题在于将 []string 转换为 bytes.buffer 以便 gob 能够从中读取 io.read 。听起来很基本，但到目前为止我已经尝试了很多方法但没有成功。这是当前的一个。问题显然是 gob 产生“缓冲区中的额外数据”，而发送之前和接收之后的数据似乎相同，如 print 语句所示。

有没有更简单、更可行的方法来解决这个问题？如果您有 zmq4，则下面的代码是独立的并且应该执行。

package main

import (
    "bytes"
    "encoding/gob"
    "fmt"
    "sync"
    "time"

    zmq "github.com/pebbe/zmq4"
)

type logentry struct {
    errid  int
    name   string
    level  string
    logstr string
}

// the client task
// --------------------------------------------------------------
func client_task(s string) {
    var mu sync.mutex
    client, _ := zmq.newsocket(zmq.dealer)
    defer client.close()
    client.setidentity(s)
    client.connect("tcp://localhost:5570")

    go func() {

        alogentry := &logentry{
            errid:  1,
            name:   "client",
            logstr: "log msg sent",
        }

        for request_nbr := 1; true; request_nbr++ {

            var network bytes.buffer
            enc := gob.newencoder(&network)
            err := enc.encode(alogentry)
            if err != nil {
                fmt.println("encode error:", err)
            }

            // early decode test - this will influence subsequent gob
            // behaviour so leave commented when caring about the sent
            // data
            // dec := gob.newdecoder(&network)
            // var alogentry2 *logentry
            // err = dec.decode(&alogentry2)
            // if err != nil {
            //  fmt.printf("client_task(decode error) : %s\n\n", err)
            // }
            // fmt.printf("client_task(test decode) %+v\n\n", alogentry)

            mu.lock()
            // replaced length by bytes buffer method: 91
            fmt.printf("client_task(len) : %d\n\n", network.len())
            fmt.printf("client_task(network) : %v\n\n", network)

            client.sendmessage(network, 0)
            mu.unlock()
            time.sleep(5 * time.second)
        }
    }()

    // pause to allow server
    for {
        time.sleep(100 * time.millisecond)
    }
}

// the server task
// --------------------------------------------------------------
func server_task() {

    frontend, _ := zmq.newsocket(zmq.router)
    defer frontend.close()
    frontend.bind("tcp://*:5570")

    for {
        msg, _ := frontend.recvmessage(0)
        // added error reporting - does not report any error
        // err does never get filled in here, never an error could get reported here
        if err != nil {
            fmt.printf("recv error: %s", err)
        }

        // using writestring to write the content portion of the
        // received message to the bytes.buffer for gob to process
        var network bytes.buffer
        dec := gob.newdecoder(&network)
        network.writestring(msg[1])

        // added length of the bytes buffer: 285
        // before sending the bytes buffer is: 91
        // more than just msg[1] is written into the buffer ?
        fmt.printf("server_task(len): %d\n\n", network.len())
        fmt.printf("server_task(msg[1]) : %s\n\n", msg[1])

        var alogentry *logentry
        err := dec.decode(&alogentry)
        if err != nil {
            fmt.printf("server_task(decode error) : %s\n\n", err)
        }

        fmt.printf("server_task(alogentry) %+v\n\n", alogentry)
    }
}

func main() {
    defer fmt.println("main() done")

    go client_task("1")
    go server_task()

    //  run for 5 seconds then quit
    time.sleep(5 * time.second)
}

打印语句在客户端显示：

client_task(network) : {[62 255 129 3 1 1 8 76 111 103 69 110 116 114 121 1 255 130 0 1 4 1 5 69 114 114 73 68 1 4 0 1 4 78 97 109 101 1 12 0 1 5 76 101 118 101 108 1 12 0 1 6 76 111 103 83 116 114 1 12 0 0 0 27 255 130 1 2 1 6 67 108 105 101 110 116 2 12 76 111 103 32 109 115 103 32 115 101 110 116 0] 0 0}

在服务器端：

server_task(msg[1]) : {[62 255 129 3 1 1 8 76 111 103 69 110 116 114 121 1 255 130 0 1 4 1 5 69 114 114 73 68 1 4 0 1 4 78 97 109 101 1 12 0 1 5 76 101 118 101 108 1 12 0 1 6 76 111 103 83 116 114 1 12 0 0 0 27 255 130 1 2 1 6 67 108 105 101 110 116 2 12 76 111 103 32 109 115 103 32 115 101 110 116 0] 0 0}

这看起来几乎是一样的。

结果是：

server_task(DECODE ERROR) : extra data in buffer
server_task(aLogEntry) <nil>

解决方案

观察

zeromq 本机 api 定义了此属性：

解决方案

要么开始使用足够正确的push/pull可扩展的正式原型，而不是（对于您的用例来说过于复杂）dealer/router，
或者
可能依赖于假设您的 router-node 永远不会 .recvmessage( 0 ) 与其他内部多部分结构，但只有一个，匹配
[ | 模板] <[网络]-有效负载> [| ... ] ]
这不能得到强有力的保证，可以吗？

虽然不确定，pebbe 的 zeromq 的 zmq4 go-wrapper 如何尝试或不实现所有本机 api 功能和/或处理中的潜在差异（无需任何用户级应用程序干预即可自动化？） 读取所有多部分组件并处理一次性和/或 null 终止字符串的处理，这可能会在 .decode() 方法内部发生冲突。

最后但并非最不重要的一点是，如果您的代码依赖于 msg[1] 来在内部拥有有效且符合所有约定的有效负载，如果我没有忽略一些低级黑客，我不知道，我看不到明确的处理情况，当始发端（ dealer ）尚未将任何此类新消息传递到消费者端（ router ），但 .recvmessage( 0 ) 方法填充 msg 并继续（使用空 msgzqben dczqb）朝向 .decode() 方法，由于明显的原因，它必须在空或格式错误的 msg 上失败，不是吗？

我肯定会从 push/pull 替代品开始，它不会在交付端注入预先准备的、现在的多帧组合，以及 routing_id 和相关风险。

如果 pull-rxqueue 缓冲区内没有等待消息，则从 .recvmessage() 方法返回的非阻塞模式仍将在用空数据填充 msg 时发生冲突，这仍将导致 .decode 出现恐慌()-方法。

如果 .recvmessage( 0 ) 方法调用实际上表现出阻塞模式接收，则如果要从错误根源中完全排除 zeromq，则应该更加小心错误状态检测和处理——原因分析。对所有已部署的 zeromq 资源进行更多自我防御的 .setsockopt()-setups （ zmq_linger 等）也将提高稳健性和易出错级别，即针对崩溃的应用程序可能造成任何伤害的情况正在生产中。

您可以尝试重现错误：here，遗憾的是 ide 错过了 zeromq 部分。

golang.org site 失败：

go: finding module for package github.com/pebbe/zmq4
go: downloading github.com/pebbe/zmq4 v1.2.0
go: found github.com/pebbe/zmq4 in github.com/pebbe/zmq4 v1.2.0
# pkg-config --cflags  -- libzmq
pkg-config: exec: "pkg-config": executable file not found in $PATH

Go build failed.

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~