{"id":43130,"date":"2024-12-01T13:05:57","date_gmt":"2024-12-01T05:05:57","guid":{"rendered":"https:\/\/fwq.ai\/blog\/43130\/"},"modified":"2024-12-01T13:05:57","modified_gmt":"2024-12-01T05:05:57","slug":"%e4%b8%8e-grpc-%e5%ae%a2%e6%88%b7%e7%ab%af%e9%87%8d%e6%96%b0%e8%bf%9e%e6%8e%a5%e7%9a%84%e6%ad%a3%e7%a1%ae%e6%96%b9%e6%b3%95","status":"publish","type":"post","link":"https:\/\/fwq.ai\/blog\/43130\/","title":{"rendered":"\u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5"},"content":{"rendered":"

<\/b> <\/p>\n

\u5f53\u524d\u4f4d\u7f6e\uff1a ><\/span> ><\/span> ><\/span> ><\/span> \u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5<\/span><\/p>\n

\u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5<\/h1>\n

\u6765\u6e90\uff1astackoverflow<\/span>
\n2024-04-29 20:54:36<\/span>
\n<\/i>0\u6d4f\u89c8<\/span>
\n<\/i>\u6536\u85cf<\/span> <\/p>\n

\u5927\u5bb6\u597d\uff0c\u4eca\u5929\u672c\u4eba\u7ed9\u5927\u5bb6\u5e26\u6765\u6587\u7ae0\u300a\u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5\u300b<\/span>\uff0c\u6587\u4e2d\u5185\u5bb9\u4e3b\u8981\u6d89\u53ca\u5230<\/span><\/span>\uff0c\u5982\u679c\u4f60\u5bf9Golang<\/span>\u65b9\u9762\u7684\u77e5\u8bc6\u70b9\u611f\u5174\u8da3\uff0c\u90a3\u5c31\u8bf7\u5404\u4f4d\u670b\u53cb\u7ee7\u7eed\u770b\u4e0b\u53bb\u5427~\u5e0c\u671b\u80fd\u771f\u6b63\u5e2e\u5230\u4f60\u4eec\uff0c\u8c22\u8c22\uff01<\/p>\n

\u95ee\u9898\u5185\u5bb9
\n <\/p>\n

\u6211\u6709\u4e00\u4e2a go grpc \u5ba2\u6237\u7aef\u8fde\u63a5\u5230\u5728 k8s \u96c6\u7fa4\u4e2d\u7684\u4e0d\u540c pod \u4e2d\u8fd0\u884c\u7684 grpc \u670d\u52a1\u5668\u3002<\/p>\n

\u5b83\u8fd0\u884c\u826f\u597d\uff0c\u53ef\u4ee5\u63a5\u6536\u548c\u5904\u7406\u8bf7\u6c42\u3002<\/p>\n

\u6211\u73b0\u5728\u60f3\u77e5\u9053\u5728 grpc \u670d\u52a1\u5668 pod \u88ab\u56de\u6536\u7684\u60c5\u51b5\u4e0b\u5982\u4f55\u6700\u597d\u5730\u5b9e\u73b0\u5f39\u6027\u3002<\/p>\n

\u636e\u6211\u6240\u77e5\uff0cclientconn.go \u4ee3\u7801\u5e94\u8be5\u81ea\u52a8\u5904\u7406\u91cd\u65b0\u8fde\u63a5\uff0c\u4f46\u6211\u5c31\u662f\u65e0\u6cd5\u8ba9\u5b83\u5de5\u4f5c\uff0c\u6211\u62c5\u5fc3\u6211\u7684\u5b9e\u73b0\u5728\u7b2c\u4e00\u4e2a\u5b9e\u4f8b\u4e2d\u662f\u4e0d\u6b63\u786e\u7684\u3002<\/p>\n

\u4ece main \u8c03\u7528\u4ee3\u7801\uff1a<\/p>\n

go func() {     \n        if err := grpcclient.processrequests(); err != nil {\n            log.error(\"error while processing requests\")\n            \/\/do something here??\n        }\n    }()<\/pre>\n
\u6211\u5728 grpcclient \u5305\u88c5\u5668\u6a21\u5757\u4e2d\u7684\u4ee3\u7801\uff1a<\/p>\n
func (grpcclient *grpcclient) processrequests() error {\n    defer grpcclient.close()    \n\n    for {\n        request, err := reqclient.stream.recv()\n        log.info(\"request received\")\n        if err == io.eof {          \n            break\n        }\n        if err != nil {\n            \/\/when pod is recycled, this is what's hit with err:\n            \/\/rpc error: code = unavailable desc = transport is closing\"\n\n            \/\/what is the correct pattern for recovery here so that we can await connection\n            \/\/and continue processing requests once more?\n            \/\/should i return err here and somehow restart the processrequests() go routine in the \n            \/\/main funcition?\n            break\n            \n        } else {\n            \/\/the happy path\n            \/\/code block to process any requests that are received\n        }\n    }\n\n    return nil\n}\n\nfunc (reqclient *requestclient) close() {\n\/\/this is called soon after the conneciton drops\n        reqclient.conn.close()\n}<\/pre>\n
\u7f16\u8f91\uff1a \u827e\u654f\u00b7\u62c9\u83b1\u6258\u7ef4\u5947\uff08emin laletovic\uff09\u5728\u4e0b\u9762\u4f18\u96c5\u5730\u56de\u7b54\u4e86\u6211\u7684\u95ee\u9898\uff0c\u5e76\u4e14\u5927\u90e8\u5206\u5185\u5bb9\u90fd\u5f97\u5230\u4e86\u89e3\u7b54\u3002 \u6211\u5fc5\u987b\u5bf9 waituntilready \u51fd\u6570\u8fdb\u884c\u4e00\u4e9b\u66f4\u6539\uff1a<\/p>\n
func (grpcclient *gRPCClient) waitUntilReady() bool {\nctx, cancel := context.WithTimeout(context.Background(), 300*time.Second) \/\/define how long you want to wait for connection to be restored before giving up\ndefer cancel()\n\ncurrentState := grpcclient.conn.GetState()\nstillConnecting := true\n\nfor currentState != connectivity.Ready && stillConnecting {\n    \/\/will return true when state has changed from thisState, false if timeout\n    stillConnecting = grpcclient.conn.WaitForStateChange(ctx, currentState)\n    currentState = grpcclient.conn.GetState()\n    log.WithFields(log.Fields{\"state: \": currentState, \"timeout\": timeoutDuration}).Info(\"Attempting reconnection. State has changed to:\")\n}\n\nif stillConnecting == false {\n    log.Error(\"Connection attempt has timed out.\")\n    return false\n}\n\nreturn true\n}<\/pre>\n
 <\/p>\n
\u89e3\u51b3\u65b9\u6848<\/h2>\n
 <\/p>\n
rpc \u8fde\u63a5\u7531 clientconn.go<\/code> \u81ea\u52a8\u5904\u7406\uff0c\u4f46\u8fd9\u5e76\u4e0d\u610f\u5473\u7740\u6d41\u4e5f\u4f1a\u81ea\u52a8\u5904\u7406\u3002<\/p>\n
\u6d41\u4e00\u65e6\u65ad\u5f00\uff0c\u65e0\u8bba\u662f\u7531\u4e8e rpc \u8fde\u63a5\u4e2d\u65ad\u8fd8\u662f\u5176\u4ed6\u539f\u56e0\uff0c\u90fd\u65e0\u6cd5\u81ea\u52a8\u91cd\u65b0\u8fde\u63a5\uff0c\u4e00\u65e6 rpc \u8fde\u63a5\u6062\u590d\uff0c\u60a8\u9700\u8981\u4ece\u670d\u52a1\u5668\u83b7\u53d6\u65b0\u7684\u6d41\u3002<\/p>\n
\u7b49\u5f85 rpc \u8fde\u63a5\u5904\u4e8e ready<\/code> \u72b6\u6001\u5e76\u5efa\u7acb\u65b0\u6d41\u7684\u4f2a\u4ee3\u7801\u53ef\u80fd\u5982\u4e0b\u6240\u793a\uff1a<\/p>\n
func (grpcclient *grpcclient) processrequests() error {\n    defer grpcclient.close()    \n    \n    go grpcclient.process()\n    for {\n      select {\n        case <- grpcclient.reconnect:\n           if !grpcclient.waituntilready() {\n             return errors.new(\"failed to establish a connection within the defined timeout\")\n           }\n           go grpcclient.process()\n        case <- grpcclient.done:\n          return nil\n      }\n    }\n}\n\nfunc (grpcclient *grpcclient) process() {\n    reqclient := getstream() \/\/always get a new stream\n    for {\n        request, err := reqclient.stream.recv()\n        log.info(\"request received\")\n        if err == io.eof {          \n            grpcclient.done <- true\n            return\n        }\n        if err != nil {\n            grpcclient.reconnect <- true\n            return\n            \n        } else {\n            \/\/the happy path\n            \/\/code block to process any requests that are received\n        }\n    }\n}\n\nfunc (grpcclient *grpcclient) waituntilready() bool {\n  ctx, cancel := context.withtimeout(context.background(), 60*time.second) \/\/define how long you want to wait for connection to be restored before giving up\n  defer cancel()\n  return grpcclient.conn.waitforstatechange(ctx, conectivity.ready)\n}<\/pre>\n
\u7f16\u8f91\uff1a<\/strong><\/p>\n
\u91cd\u65b0\u5ba1\u89c6\u4e0a\u9762\u7684\u4ee3\u7801\uff0c\u5e94\u8be5\u7ea0\u6b63\u4e00\u4e9b\u9519\u8bef\u3002 waitforstatechange<\/code> \u51fd\u6570\u7b49\u5f85\u8fde\u63a5\u72b6\u6001\u4ece\u4f20\u9012\u72b6\u6001\u66f4\u6539\uff0c\u5b83\u4e0d\u7b49\u5f85\u8fde\u63a5\u66f4\u6539\u4e3a\u4f20\u9012\u72b6\u6001\u3002<\/p>\n
\u6700\u597d\u8ddf\u8e2a\u5f53\u524d\u8fde\u63a5\u72b6\u6001\uff0c\u5982\u679c\u901a\u9053\u7a7a\u95f2\uff0c\u5219\u4f7f\u7528 connect<\/code> \u51fd\u6570\u8fdb\u884c\u8fde\u63a5\u3002<\/p>\n
func (grpcclient *grpcclient) processrequests() error {\n        defer grpcclient.close()    \n        \n        go grpcclient.process()\n        for {\n          select {\n            case <- grpcclient.reconnect:\n               if !grpcclient.isreconnected(1*time.second, 60*time.second) {\n                 return errors.new(\"failed to establish a connection within the defined timeout\")\n               }\n               go grpcclient.process()\n            case <- grpcclient.done:\n              return nil\n          }\n        }\n}\n\nfunc (grpcclient *grpcclient) isreconnected(check, timeout time.duration) bool {\n  ctx, cancel := context.context.withtimeout(context.background(), timeout)\n  defer cancel()\n  ticker := time.newticker(check)\n\n  for{\n    select {\n      case <- ticker.c:\n        grpcclient.conn.connect()\n \n        if grpcclient.conn.getstate() == connectivity.ready {\n          return true\n        }\n      case <- ctx.done():\n         return false\n    }\n  }\n}<\/pre>\n
\u5f53grpc\u8fde\u63a5\u5173\u95ed\u65f6\uff0cgrpc\u5ba2\u6237\u7aef\u8fde\u63a5\u7684\u72b6\u6001\u5c06\u4e3a idle<\/code> \u6216 transient_failure<\/code>\u3002\u4ee5\u4e0b\u662f\u6211\u7684 grpc \u53cc\u5411\u6d41\u5f0f\u4f20\u8f93\u81ea\u5b9a\u4e49\u91cd\u65b0\u8fde\u63a5\u673a\u5236\u7684\u793a\u4f8b\u3002\u9996\u5148\uff0c\u6211\u6709\u4e00\u4e2a for \u5faa\u73af\u6765\u4fdd\u6301\u91cd\u65b0\u8fde\u63a5\uff0c\u76f4\u5230 grpc \u670d\u52a1\u5668\u542f\u52a8\uff0c\u5728\u8c03\u7528 conn.connect()<\/code> \u540e\u72b6\u6001\u5c06\u53d8\u4e3a\u5c31\u7eea\u72b6\u6001\u3002<\/p>\n
for {\n    select {\n    case <-ctx.done():\n        return false\n    default:\n            if client.conn.getstate() != connectivity.ready {\n                client.conn.connect()\n            }\n\n            \/\/ reserve a short duration (customizable) for conn to change state from idle to ready if grpc server is up\n            time.sleep(500 * time.millisecond)\n\n            if client.conn.getstate() == connectivity.ready {\n                return true\n            }\n\n            \/\/ define reconnect time interval (backoff) or\/and reconnect attempts here\n            time.sleep(2 * time.second)\n    }\n}<\/pre>\n
\u6b64\u5916\uff0c\u8fd8\u5c06\u751f\u6210\u4e00\u4e2a goroutine \u4ee5\u6267\u884c\u91cd\u65b0\u8fde\u63a5\u4efb\u52a1\u3002\u6210\u529f\u91cd\u8fde\u540e\uff0c\u4f1a\u751f\u6210\u53e6\u4e00\u4e2agoroutine\u6765\u76d1\u542cgrpc\u670d\u52a1\u5668\u3002<\/p>\n
for {\n    select {\n    case <-ctx.done():\n        return\n    case <-reconnectch:\n        if client.conn.getstate() != connectivity.ready && *isconnectedwebsocket {\n            if o.waituntilready(client, isconnectedwebsocket, ctx) {\n                err := o.generatenewprocessorderstream(client, ctx)\n                if err != nil {\n                    logger.logger.error(\"failed to establish stream connection to grpc server ...\")\n                }\n\n                \/\/ re-listening server side streaming\n                go o.listenprocessorderserverside(client, reconnectch, ctx, isconnectedwebsocket)\n            }\n        }\n    }\n}<\/pre>\n
\u8bf7\u6ce8\u610f\uff0c\u76d1\u542c\u4efb\u52a1\u662f\u7531\u53e6\u4e00\u4e2a goroutine \u5e76\u53d1\u5904\u7406\u7684\u3002<\/p>\n
\/\/ listening server side streaming\ngo o.listenProcessOrderServerSide(client, reconnectCh, websocketCtx, isConnectedWebSocket)<\/pre>\n
\u60a8\u53ef\u4ee5\u67e5\u770b\u6211\u7684\u4ee3\u7801\u793a\u4f8b \u3002\u5e0c\u671b\u8fd9\u4f1a\u6709\u6240\u5e2e\u52a9\u3002<\/p>\n
\u56fe\u7247\u6765\u6e90\uff1a\u827e\u654f\u00b7\u62c9\u83b1\u6258\u7ef4\u5947<\/p>\n
\u597d\u4e86\uff0c\u672c\u6587\u5230\u6b64\u7ed3\u675f\uff0c\u5e26\u5927\u5bb6\u4e86\u89e3\u4e86\u300a\u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5\u300b\uff0c\u5e0c\u671b\u672c\u6587\u5bf9\u4f60\u6709\u6240\u5e2e\u52a9\uff01\u5173\u6ce8\u7c73\u4e91\u516c\u4f17\u53f7\uff0c\u7ed9\u5927\u5bb6\u5206\u4eab\u66f4\u591aGolang\u77e5\u8bc6\uff01<\/p>\n","protected":false},"excerpt":{"rendered":"
\u5f53\u524d\u4f4d\u7f6e\uff1a > > > > \u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5 \u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5 \u6765\u6e90\uff1astackoverflow 2024-04-29 20:54:36 0\u6d4f\u89c8 \u6536\u85cf \u5927\u5bb6\u597d\uff0c\u4eca\u5929\u672c\u4eba\u7ed9\u5927\u5bb6\u5e26\u6765\u6587\u7ae0\u300a\u4e0e gRPC \u5ba2\u6237\u7aef\u91cd\u65b0\u8fde\u63a5\u7684\u6b63\u786e\u65b9\u6cd5\u300b\uff0c\u6587\u4e2d\u5185\u5bb9\u4e3b\u8981\u6d89\u53ca\u5230\uff0c\u5982\u679c\u4f60\u5bf9Golang\u65b9\u9762\u7684\u77e5\u8bc6\u70b9\u611f\u5174\u8da3\uff0c\u90a3\u5c31\u8bf7\u5404\u4f4d\u670b\u53cb\u7ee7\u7eed\u770b\u4e0b\u53bb\u5427~\u5e0c\u671b\u80fd\u771f\u6b63\u5e2e\u5230\u4f60\u4eec\uff0c\u8c22\u8c22\uff01 \u95ee\u9898\u5185\u5bb9 \u6211\u6709\u4e00\u4e2a go grpc \u5ba2\u6237\u7aef\u8fde\u63a5\u5230\u5728 k8s \u96c6\u7fa4\u4e2d\u7684\u4e0d\u540c pod \u4e2d\u8fd0\u884c\u7684 grpc \u670d\u52a1\u5668\u3002 \u5b83\u8fd0\u884c\u826f\u597d\uff0c\u53ef\u4ee5\u63a5\u6536\u548c\u5904\u7406\u8bf7\u6c42\u3002 \u6211\u73b0\u5728\u60f3\u77e5\u9053\u5728 grpc \u670d\u52a1\u5668 pod \u88ab\u56de\u6536\u7684\u60c5\u51b5\u4e0b\u5982\u4f55\u6700\u597d\u5730\u5b9e\u73b0\u5f39\u6027\u3002 \u636e\u6211\u6240\u77e5\uff0cclientconn.go \u4ee3\u7801\u5e94\u8be5\u81ea\u52a8\u5904\u7406\u91cd\u65b0\u8fde\u63a5\uff0c\u4f46\u6211\u5c31\u662f\u65e0\u6cd5\u8ba9\u5b83\u5de5\u4f5c\uff0c\u6211\u62c5\u5fc3\u6211\u7684\u5b9e\u73b0\u5728\u7b2c\u4e00\u4e2a\u5b9e\u4f8b\u4e2d\u662f\u4e0d\u6b63\u786e\u7684\u3002 \u4ece main \u8c03\u7528\u4ee3\u7801\uff1a go func() { if err := grpcclient.processrequests(); err != nil { log.error(“error while processing […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[17],"tags":[],"class_list":["post-43130","post","type-post","status-publish","format-standard","hentry","category-docker"],"_links":{"self":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/posts\/43130","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/comments?post=43130"}],"version-history":[{"count":0,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/posts\/43130\/revisions"}],"wp:attachment":[{"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/media?parent=43130"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/categories?post=43130"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/fwq.ai\/blog\/wp-json\/wp\/v2\/tags?post=43130"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}