当前位置：首页 > 文章列表 > Golang > Go问答 > ZeroMQ Golang 在 Fargate 上的应用

ZeroMQ Golang 在 Fargate 上的应用

来源：stackoverflow 2024-03-22 09:42:50 0浏览收藏

推广推荐

支持 PC / 移动端，安全直达

在 Fargate 中使用 ZeroMQ Go 应用程序时，可能会遇到连接问题。通过分步分析问题，我们可以确定缺少跨服务连接，从而导致容器无法相互通信。服务之间的连接需要正确的服务发现机制，以便容器能够通过名称和端口进行调用，即使它们位于不同的服务中。

问题内容

我尝试在以 awsvpc 模式在 fargate 上运行的 ecs 中使用 zeromq。我有 2 个不同的服务，每个服务都在启用服务发现的情况下运行自己的任务。

我在名为代理的微服务中创建路由器和经销商。

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

back, _ := zmq.newsocket(zmq.dealer)
defer back.close()
back.bind("tcp://*:4080")

然后，我将这 2 个套接字添加到轮询器中，并有一个等待消息的 for 循环。

我有一个单独的微服务，它连接到套接字并尝试向经销商发送消息。我已经设置了服务发现，所以我假设我连接的地址是：

“tcp://经纪人:4070”

下面是“servicea”的代码

func new(zmq models.zmq) *requester {
    s, err := zmq.newsocket(zmq.req)
    if err != nil {
        log.fatalln("shareholder/requester zmq.newsocket", err)
    }
    p := zmq.newpoller()
    p.add(s, zmq.pollin)

    log.println("requester", zmq.req)
    err = s.connect("tcp://broker:4070")
    if err != nil {
        log.print(fmt.errorf("err is %w", err))
    }

    req := &requester{
        poller:  p,
        retries: 2,
        socket:  s,
        timeout: time.duration(time.minute),
    }
    runtime.setfinalizer(req, (*requester).close)
    return req
}

然后我使用上面的代码通过我的套接字连接发送消息

_, err := r.Socket.SendMessage(req)

但是，我的代理服务中从未收到我的消息。我可以使用我在服务发现期间注册的主机名来访问网络上的 rest api，我在 fargate/ecs/zeromq 中缺少什么吗？？？

解决方案

问：“我在 fargate/ecs/zeromq 中是否缺少某些东西？？？”

也许是，也许不是。

让我们以结构化的方式开始深入探究根本原因：

第 0 步：`broker` 服务节点

提到要使用zeromq，所以我们就从这一点开始。假设您的选择是在地址 ( *:4070 ) 上使用 dealer 的访问点，并在地址 ( *:4080 ) 上使用 router 的访问点，并且两者都使用 .bind() 方法来激活tcp://-broker内的传输类-微服务节点，我们的下一步是验证该节点是否以及如何实际上可见对于世界其他地方。

所以，让它运行。

第 1 步：视线测试

这是测试的第一步 - broker-节点（无论其实现如何）实际上对“目标受众”可见吗？如果没有，在 zeromq 或其他框架内没有太多可做的，但你的任务是获取地址、l1 信号互连、l2-arp/rarp mac 检测/映射、l3 路由权限/访问列表/filters/xlations/etc，（动态）dns 更新和所有其他配置更新，以便您能够让世界其他地区（选择性部分）看到并更接近成功 .connect()

$ #                                 is it l3-(in)-visible # a [ pass ] | [ fail ]
$ ping <_a_broker_node_assumed_tcp/ip_address>            # a [ pass ] | [ fail ]

第 2 步：端口号 rto 测试

$ #                                                  4070 # a [ pass ] | [ fail ]
$ netcat -vz <_a_broker_node_visible_tcp/ip_address> 4070 # a [ pass ] | [ fail ]
$ ######
$ # or :
$ ######
$ telnet     <_a_broker_node_visible_tcp/ip_address> 4070 # a [ pass ] | [ fail ]
trying 
connected to 
escape character is '^]'.
https://<_a_broker_node_visible_tcp/ip_address>:4070
http/1.1 400 bad request
server: nginx
date: mon, 03 may 2020 18:14:54 gmt
content-type: text/html
content-length: 150
connection: close

<html>
<head><title>400 bad request</title></head>
<body>
<center><h1>400 bad request</h1></center>
<hr><center>nginx</center>
</body>
</html>
connection closed by foreign host.
$
$ //                                             4080 // a [ pass ] | [ fail ]
$ telnet <_a_broker_node_visible_tcp/ip_address> 4080 // a [ pass ] | [ fail ]

第 3 步：本地消息发送的 rto 测试

替换 req/router-可扩展的正式通信原型模式中相当复杂的领域，让我们使用简单的 push/pull 消息传递测试进行测试，该测试（出于明显的原因）与预期匹配用于发送消息：

package main

import (
    zmq "github.com/pebbe/zmq4"
    "log"
    "fmt"
    "time"
    ...
)

func PushTASK() {

    aCtx, err    := zmq.NewContext()
    if err != nil {
        log.Fatalln( "__NACK: aCtx instantiation failed in zmq.NewContext()",
                      err )
    }

    aPusher, err := aCtx.NewSocket( zmq.PUSH )
    if err != nil {
        log.Fatalln( "__NACK: aPusher instantiation failed in aCtxNewSocket()",
                      err )
    }

    err = aPusher.SetLinger( 0 )
    if err != nil {
        log.Fatalln( "__NACK: aPusher instance failed to .SetLinger()",
                      err  )
    }

    err = aPusher.SetConflate( true )
    if err != nil {
        log.Fatalln( "__NACK: aPusher instance failed to .SetConflate()",
                      err  )
    }

    log.Println( "POSACK: aPusher instantiated and about to .connect( tcp://addr:port#)" )

    err = aPusher.Connect( "tcp://broker:4070" )
    if err != nil {
        log.Print( fmt.Errorf( "__NACK: aPusher failed to .connect(): %w",
                                err )
                   )
    }

    log.Println( "POSACK: aPusher RTO and about to .SendMessage*()-loop" )

    for aPush_NUMBER := 1; aPush_NUMBER < 10000; aPush_NUMBER++ {

        err = aPusher.SendMessageDontwait( aPush_NUMBER )
        if err != nil {
              log.Print( fmt.Errorf( "__NACK: aPusher failed to .SendMessageDontwait()[%d]: %w",
                                      aPush_NUMBER,
                                      err )
                         )
        }

        time.Sleep( 0.1 * time.Second )
    }
 // ---------------------------------------------------BE NICE TO RESOURCES USED
    err = aPusher.Disconnect( "tcp://broker:4070" )
    if err != nil {
        log.Print( fmt.Errorf( "__NACK: aPusher failed to .Disconnect( tcp://addr:port ): %w",
                                err )
                   )
    }
 // ---------------------------------------------------BE NICE TO RESOURCES USED
    err = aPusher.Close()
    if err != nil {
        log.Print( fmt.Errorf( "__NACK: aPusher failed to .Close(): %w",
                                err )
                   )
    }
 // ---------------------------------------------------BE NICE TO RESOURCES USED
    err = aCtx.Term()
    if err != nil {
        log.Print( fmt.Errorf( "__NACK: aCtx failed to .Term(): %w",
                                err )
                   )
    }
 // ---------------------------------------------------WE ARE CLEAR TO TERMINATE
}