我正在嘗試在以 awsvpc 模式在 Fargate 上運行的 ECS 中使用 ZeroMQ。我有 2 個不同的服務，每個服務都運行自己的任務并啟用服務發現。我在一個名為 broker 的微服務中創建了我的 Router 和 Dealer。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://broker: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 有什么我在這里缺少的嗎？
2 回答
瀟瀟雨雨
TA貢獻1833條經驗 獲得超4個贊
問：“Fargate/ECS/ZeroMQ 有什么我在這里缺少的嗎？？？”
可能是，可能不是。
讓我們以結構化的方式開始，深入了解根本原因：
步驟 0：Broker服務節點
提到要使用 ZeroMQ，所以我們將從這一點開始。鑒于您的選擇是使用 AccessPoint 到DEALERon address( *:4070 )和 AccessPoint to ROUTERon address ( *:4080 )，并且都使用.bind()-method 來激活-Microservice 節點tcp://內的 -Transport 類Broker，我們的下一步是驗證該節點是否以及如何實際對世界其他地方可見。
所以，讓它運行。
第 1 步：視線測試
這是測試的第一步Broker--Node，無論它的實現是什么，對“目標受眾”實際上是可見的嗎？如果沒有，ZeroMQ 或其他框架內沒有太多可做的，但您的任務是獲取地址、L1 信號互連、L2-arp/rarp MAC-檢測/映射、L3-路由權限/訪問列表/filters/xlations/etc，（動態）DNS-updates 和所有其他配置更新，這樣您就可以讓世界其他地方的（選擇性部分）看到并離成功更近一步.connect()
$ # is it L3-(in)-visible # a [ PASS ] | [ FAIL ]
$ ping <_a_Broker_Node_Assumed_TCP/IP_Address> # a [ PASS ] | [ FAIL ]
第 2 步：端口號 RTO-Test
$ # 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-Scalable Formal Communications Archetype Pattern 領域，讓我們用一個簡單的PUSH/PULL-message 傳遞測試進行測試，它（出于顯而易見的原因）與發送消息的預期用途相匹配：
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
}
第 4 步：遠程消息接收的 RTO 測試
如果[ PASS ] | [ FAIL ]-tests 都沒有崩潰，下一步是反映PUSH"remote" 的 -side 概念Broker，是的，重寫它以使用PULL-side 并部署它以查看是否也沒有崩潰以及是否消息在仍在運行或重新運行第 3 步時按應有的方式到達。
第 5 步：享受 ZeroMQ 的強大功能
一旦上述所有測試確實完成了[ PASS ]，您不僅可以確定 ZeroMQ 不是阻礙因素，而且還可以將部署的原則增強到任何進一步的用例場景中，給定 L1-/L2-/L3-/ZeroMQ-服務以正確和可驗證的方式落實到位。
- 2 回答
- 0 關注
- 196 瀏覽
添加回答
舉報
0/150
提交
取消