我的Go gRPC之旅、02 四種通信模式

来源:https://www.cnblogs.com/linxiaoxu/archive/2022/09/20/16709673.html
-Advertisement-
Play Games

我的gRPC之旅。本節簡單介紹gRPC的四種通信模式。簡單通信模式、服務端流通信模式、客戶端流通信模式、雙向流通信模式。 ...


藉助gRPC我們可以實現不同進程間通信模式(也稱RPC風格)。

repeated 關鍵字

message Order {
    string id = 1;
    repeated string items = 2;
    string description = 3;
    float price = 4;
    string destination = 5;
}

使用repeated表明這個欄位在消息中可以重覆出現多次,包括0次。編譯成go,結構體會表示成一個切片。

一元RPC模式

01 初識gRPC,感受gRPC的強大魅力 - 小能日記 - 博客園

一元RPC模式也被稱為簡單RPC模式。在該模式中,當客戶端調用伺服器端的遠程方法時,客戶端發送請求至伺服器端並獲得一個響應,與響應一起發送的還有狀態細節以及trailer元數據。

   rpc addOrder(Order) returns (google.protobuf.StringValue);
   rpc getOrder(google.protobuf.StringValue) returns (Order);

編譯後

func (s *server) GetOrder(ctx context.Context, orderId *wrapper.StringValue) (*pb.Order, error) {

其中Context對象傳遞到方法中是因為其包含了一些用於控制gRPC行為的構造,比如截止時間和取消功能。

伺服器端流RPC模式

服務端在接收到客戶端的請求消息後,會發回一個響應的序列。這種多個響應所組成的序列也被稱為”流“。

在將所有的服務端響應發送完畢之後,服務端會以trailer元數據的形式將其狀態發送給客戶端,從而標記流的結束。

訂單服務的客戶端發出一個請求之後,會接收到多條響應消息。

    rpc searchOrders(google.protobuf.StringValue) returns (stream Order);

通過使用 returns (stream Order) 將返回參數指定為訂單的流。編譯後

func (s *server) SearchOrders(searchQuery *wrappers.StringValue, stream pb.OrderManagement_SearchOrdersServer) error {
	for key, order := range orderMap {
		for _, itemStr := range order.Items {
			if strings.Contains(itemStr, searchQuery.Value) {
				err := stream.Send(&order) 
                // 需要處理將消息以流的形式發送給客戶端的過程中可能出現的錯誤
				if err != nil {
					return fmt.Errorf("error sending message to stream : %v", err)
				}
				log.Print("Matching Order Found : " + key)
				break
			}
		}
	}
	return nil
}

pb.OrderManagement_SearchOrdersServer 是服務端流的寫入對象,可以寫入多個響應。

客戶端代碼使用Recv方法從客戶端流中檢索消息,並且持續檢索,直到流結束為止,即 io.EOF

searchStream, 	_ := client.SearchOrders(ctx, &wrapper.StringValue{Value: "Google"})
for {
    searchOrder, err := searchStream.Recv()
    if err == io.EOF {
        log.Print("EOF")
        break
    }

    if err == nil {
        log.Print("Search Result : ", searchOrder)
    }
}

客戶端流RPC模式

客戶端會發送多個請求給服務端。服務端可以隨時結束接收或接收所有消息後再發送響應。

    rpc updateOrders(stream Order) returns (google.protobuf.StringValue);

編譯為

func (s *server) UpdateOrders(stream pb.OrderManagement_UpdateOrdersServer) error {
	ordersStr := "Updated Order IDs : "
	for {
		order, err := stream.Recv()
		if err == io.EOF {
            // 客戶端已發送完畢,伺服器可以響應
			return stream.SendAndClose(&wrapper.StringValue{Value: "Orders processed " + ordersStr})
		}

		if err != nil {
			return err
		}
		orderMap[order.Id] = *order

		log.Printf("Order ID : %s - %s", order.Id, "Updated")
		ordersStr += order.Id + ", "
	}
}

pb.OrderManagement_UpdateOrdersServer是客戶端傳入消息流的引用對象。

服務端調用該對象的SendAndClose方法可以發送響應,同時標記伺服器端消息終結了流。

客戶端調用對象的CloseAndRecv方法可以關閉流並接收響應。

	updateStream, err := client.UpdateOrders(ctx)

	if err != nil {
		log.Fatalf("%v.UpdateOrders(_) = _, %v", client, err)
	}

	if err := updateStream.Send(&updOrder1); err != nil {
		log.Fatalf("%v.Send(%v) = %v", updateStream, updOrder1, err)
	}

	if err := updateStream.Send(&updOrder2); err != nil {
		log.Fatalf("%v.Send(%v) = %v", updateStream, updOrder2, err)
	}

	if err := updateStream.Send(&updOrder3); err != nil {
		log.Fatalf("%v.Send(%v) = %v", updateStream, updOrder3, err)
	}
	// 結束流並等待服務端響應
	updateRes, err := updateStream.CloseAndRecv()
	if err != nil {
		log.Fatalf("%v.CloseAndRecv() got error %v, want %v", updateStream, err, nil)
	}
	log.Printf("Update Orders Res : %s", updateRes)

雙向流模式

雙向流模式中,客戶端以消息流的形式發送請求到服務端,服務端也以消息流的形式進行響應。調用必須由客戶端發起。流的操作完全獨立,客戶端和服務端可以按照任意順序進行讀取和寫入。

    rpc processOrders(stream google.protobuf.StringValue) returns (stream CombinedShipment);

一旦調用RPC方法,那麼無論是客戶端還是服務端,都可以在任意時間發送消息。這也包括來自任意一段的流結束標記。編譯後

func (s *server) ProcessOrders(stream pb.OrderManagement_ProcessOrdersServer) error {
	...
}

pb.OrderManagement_ProcessOrdersServer 是客戶端和伺服器端之間消息流的對象引用。既可以Recv方法讀取,也可以Send方法寫入。

客戶端代碼中可開啟兩個線程分別用於發送消息流和讀取消息流。調用流引用對象的CloseSend方法可以關閉當前流並通知另一端,但另一端並未關閉,還可以發送數據。

	...
	streamProcOrder, err := client.ProcessOrders(ctx)
	if err != nil {
		log.Fatalf("%v.ProcessOrders(_) = _, %v", client, err)
	}

	if err := streamProcOrder.Send(&wrapper.StringValue{Value:"102"}); err != nil {
		log.Fatalf("%v.Send(%v) = %v", client, "102", err)
	}

	if err := streamProcOrder.Send(&wrapper.StringValue{Value:"103"}); err != nil {
		log.Fatalf("%v.Send(%v) = %v", client, "103", err)
	}

	if err := streamProcOrder.Send(&wrapper.StringValue{Value:"104"}); err != nil {
		log.Fatalf("%v.Send(%v) = %v", client, "104", err)
	}

	channel := make(chan struct{})
	go asncClientBidirectionalRPC(streamProcOrder, channel)
	time.Sleep(time.Millisecond * 1000)

	if err := streamProcOrder.Send(&wrapper.StringValue{Value:"101"}); err != nil {
		log.Fatalf("%v.Send(%v) = %v", client, "101", err)
	}
	if err := streamProcOrder.CloseSend(); err != nil {
		log.Fatal(err)
	}
// 用channel保證main在讀取消息流的go程結束後再結束
	channel <- struct{}{}
}
func asncClientBidirectionalRPC(streamProcOrder pb.OrderManagement_ProcessOrdersClient, c chan struct{}) {
	for {
		combinedShipment, errProcOrder := streamProcOrder.Recv()
		if errProcOrder == io.EOF {
			break
		}
		log.Printf("Combined shipment : ", combinedShipment.OrdersList)
	}
	<-c
}

您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 在面向對象的程式設計中,模塊之間交互採用介面編程,通常情況下調用方不需要知道被調用方的內部實現細節,因為一旦涉及到了具體實現,如果需要換一種實現就需要修改代碼,這違反了程式設計的"開閉原則"。所以我們一般有兩種選擇:一種是使用API(Application Programming Interface ...
  • 我的設計模式之旅。本節學習了適配器模式。從程式調用第三方庫時遇到的問題著手,思考如何讓兩個不關聯的類一起工作。並嘗試使用C#運用適配器模式解決方釘與圓孔問題。 ...
  • 上個月我寫的一篇文章《關於技術能力的思考和總結》引起了大家的關註,好多讀者的評論“以寫代想、以想促真、以講驗真”,大家的感受很深刻,基於上次的文章,這篇文章我其實更想跟大家聊聊一些常用的思考方法,思考問題的方式對了,往往可以幫助大家少走彎路。 ...
  • 現在的很多程式應用,基本上都是需要多端覆蓋,因此基於一個Web API的後端介面,來構建多端應用,如微信、H5、APP、WInForm、BS的Web管理端等都是常見的應用。本篇隨筆概括性的介紹基於HBuilderX+UniApp+ThorUI的手機端前端開發處理,總結一下開發工具的設置,以及常見的H... ...
  • 前言 秒殺請求在高度集中在某一個時間點。這樣一來,就會導致一 個特別高的流量峰值,它對資源的消耗是瞬時的 。能夠搶到商品的人數是有限的,也就是說10人和1000人發 起請求的結果都是一樣的。也就是說真正開始下單時,秒殺請求並不是越多越好。 一、秒殺中的削峰 猶豫伺服器的處理資源是恆定的,用或者不用它 ...
  • 隨著需求開發迭代,代碼庫規模逐漸變大,新的團隊成員引入等諸多因素,系統起初制定的架構規則不可避免遭到破壞。不僅僅是破壞團隊的統一開發規範,更為重要的是隨著代碼庫規模逐漸增長,大大降低系統的可維護性、擴展性,增加評審複雜度和重構成本,也最終導致團隊生產力下降以及研發成本增長。 在敏捷開發環境下,系統... ...
  • 摘要:零售企業就需要安全、可信、開放、能力強大的PaaS集成平臺支撐自身的雲業務,同樣也需要一個強大的業務系統來承載業務。 疫情又來了,買買買,趕緊囤。 這麼快沒貨了? 疫情反覆態勢之下,消費者體驗到的商品到達速度和多樣的產品選擇,以及平臺面臨的跨區調貨和各種渠道的線上流量突增等現狀,使得消費品及零 ...
  • 限流:使用Redisson的RRateLimiter進行限流 多策略:map+函數式介面優化if判斷 自定義註解 /** * aop限流註解 */ @Target({ElementType.METHOD, ElementType.TYPE}) @Retention(RetentionPolicy.R ...
一周排行
    -Advertisement-
    Play Games
  • 前言 當別人做大數據用Java、Python的時候,我使用.NET做大數據、數據挖掘,這確實是值得一說的事。 寫的並不全面,但都是實際工作中的內容。 .NET在大數據項目中,可以做什麼? 寫腳本(使用控制台程式+頂級語句) 寫工具(使用Winform) 寫介面、寫服務 使用C#寫代碼的優點是什麼? ...
  • 前言 本文寫給想學C#的朋友,目的是以儘快的速度入門 C#好學嗎? 對於這個問題,我以前的回答是:好學!但仔細想想,不是這麼回事,對於新手來說,C#沒有那麼好學。 反而學Java還要容易一些,學Java Web就行了,就是SpringBoot那一套。 但是C#方向比較多,你是學控制台程式、WebAP ...
  • 某一日晚上上線,測試同學在回歸項目黃金流程時,有一個工單項目介面報JSF序列化錯誤,馬上升級對應的client包版本,編譯部署後錯誤消失。 線上問題是解決了,但是作為程式員要瞭解問題發生的原因和本質。但這都是為什麼呢? ...
  • 本文介紹基於Python語言中TensorFlow的Keras介面,實現深度神經網路回歸的方法。 1 寫在前面 前期一篇文章Python TensorFlow深度學習回歸代碼:DNNRegressor詳細介紹了基於TensorFlow tf.estimator介面的深度學習網路;而在TensorFl ...
  • 前段時間因業務需要完成了一個工作流組件的編碼工作。藉著這個機會跟大家分享一下整個創作過程,希望大家喜歡,組件暫且命名為"easyFlowable"。 接下來的文章我將從什麼是工作流、為什麼要自研這個工作流組件、架構設計三個維度跟大家來做個整體介紹。 ...
  • 1 簡介 我們之前使用了dapr的本地托管模式,但在生產中我們一般使用Kubernetes托管,本文介紹如何在GKE(GCP Kubernetes)安裝dapr。 相關文章: dapr本地托管的服務調用體驗與Java SDK的Spring Boot整合 dapr入門與本地托管模式嘗試 2 安裝GKE ...
  • 摘要:在jvm中有很多的參數可以進行設置,這樣可以讓jvm在各種環境中都能夠高效的運行。絕大部分的參數保持預設即可。 本文分享自華為雲社區《為什麼需要對jvm進行優化,jvm運行參數之標準參數》,作者:共飲一杯無。 我們為什麼要對jvm做優化? 在本地開發環境中我們很少會遇到需要對jvm進行優化的需 ...
  • 背景 我們的業務共使用11台(阿裡雲)伺服器,使用SpringcloudAlibaba構建微服務集群,共計60個微服務,全部註冊在同一個Nacos集群 流量轉發路徑: nginx->spring-gateway->業務微服務 使用的版本如下: spring-boot.version:2.2.5.RE ...
  • 基於php+webuploader的大文件分片上傳,帶進度條,支持斷點續傳(刷新、關閉頁面、重新上傳、網路中斷等情況)。文件上傳前先檢測該文件是否已上傳,如果已上傳提示“文件已存在”,如果未上傳則直接上傳。視頻上傳時會根據設定的參數(分片大小、分片數量)進行上傳,上傳過程中會在目標文件夾中生成一個臨 ...
  • 基於php大文件分片上傳至七牛雲,使用的是七牛雲js-sdk V2版本,引入js文件,配置簡單,可以暫停,暫停後支持斷點續傳(刷新、關閉頁面、重新上傳、網路中斷等情況),可以配置分片大小和分片數量,官方文檔https://developer.qiniu.com/kodo/6889/javascrip ...