1、前言 上個星期完成了surging 的0.9.0.1 更新工作，此版本通過nuget下載引擎組件，下載後，無需通過代碼build集成，引擎會通過Sidecar模式自動掃描裝配異構組件來構建服務引擎，而這篇將介紹淺談surging服務引擎中的rabbitmq組件和容器化部署 surging源碼下載 ...
1、前言
上個星期完成了surging 的0.9.0.1 更新工作，此版本通過nuget下載引擎組件，下載後，無需通過代碼build集成，引擎會通過Sidecar模式自動掃描裝配異構組件來構建服務引擎，而這篇將介紹淺談surging服務引擎中的rabbitmq組件和容器化部署
2、Sidecar模式
比如現在比較火的Service Mesh， 談到Service Mesh，就不得不瞭解下Sidecar模式，Sidecar設計模式被越來越多的關註和採用，此模式之所以稱作Sidecar，是因為它類似於三輪摩托車上的挎鬥。 在此模式中，挎鬥附加到應用程式中，為應用程式提供支持性功能。挎鬥與應用程式具有相同的生命周期：與應用程式一起創建，一起停用。 挎鬥模式有時也稱為搭檔模式，這是一種分解模式。而surging 採用了Sidecar模式用來附加組件，而使用Sidecar模式有以下功能
- 共用存儲空間
引擎組件部署到共用的文件目錄里，服務引擎從共用的文件目錄掃描引擎組件文件。
- 共用組件和業務的配置文件
針對於組件的配置文件部署到共用的文件目錄里，服務引擎從共用的文件目錄載入文件。
- 獨立的業務服務
針對於業務可以把依賴的組件打包部署到共用的文件目錄里，服務引擎從共用的文件目錄掃描載入，從而部署成獨立的業務服務
- 內置多種協議
針對於獨立部署的業務服務，內置了多種協議，提供給服務和外部程式進行調用
模式特點
-
隔離：讓組件都能夠關註核心問題。比如eventbus、Logger、 netty 在實現功能的同時無需關註其它組件的實現而發生的衝突；
-
單一責任原則：每個組件都應該職責分開，而根據這一原則，職責應該是對應一個類、模塊或者介面，從而能夠獨立進行處理。
-
內聚性/可重用性：針對組件的特性，方法可以進行重用，從而滿足組件可持續擴展。
3、基於Event Bus 的Rabbitmq組件
surging服務引擎擴展了基於eventbus 的rabbitmq ，組件可以選擇綁定 Normal，Retry(Dead letter)，Fail ，如下圖所示。
而針對於該組件有哪些應用場景呢？
- 商品秒殺和搶購
搶購/秒殺是如今很常見的一個應用場景，在高併發的流量訪問下可以將用戶放入到搶購隊列中，購買成功則銷毀消息。
- 最終數據的一致性
在大型業務中，系統一般由多個獨立的服務組成，在分散式調用時候把消息放入到rabbitmq 隊列中，再通過消息的冪等性來解決數據的最終一致性
- 訂單失效處理
在購買商品/服務生成訂單業務中，會設定支付時間，如果一直未支付，會直接關閉訂單，而這個場景可以通過死信隊列的來解決
示例代碼
可以通過繼承BaseIntegrationEventHandler或者IIntegrationEventHandler，再通過QueueConsumer特性進行標識，具體代碼如下
[QueueConsumer("UserLoginDateChangeHandler",QueueConsumerMode.Normal)] public class UserLoginDateChangeHandler : BaseIntegrationEventHandler<UserEvent> { private readonly IUserService _userService; public UserLoginDateChangeHandler() { _userService = ServiceLocator.GetService<IUserService>("User"); } public override async Task Handle(UserEvent @event) { Console.WriteLine($"消費1。"); await _userService.Update(@event.UserId, new UserModel() { Age = @event.Age, Name = @event.Name, UserId = @event.UserId }); Console.WriteLine($"消費1失敗。"); throw new Exception(); } public override Task Handled(EventContext context) { Console.WriteLine($"調用{context.Count}次。類型:{context.Type}"); var model = context.Content as UserEvent; return Task.CompletedTask; } }
可以通過以下選項去更改配置
1 "EventBus": { 2 "EventBusConnection": "${EventBusConnection}|localhost", 3 "EventBusUserName": "${EventBusUserName}|guest",//用戶名 4 "EventBusPassword": "${EventBusPassword}|guest",//密碼 5 "VirtualHost": "${VirtualHost}|/", 6 "MessageTTL": "${MessageTTL}|30000",//消息過期時間，比如過期時間是30分鐘就是1800000 7 "RetryCount": "${RetryCount}|1",//重試次數，這裡設置的延遲隊列，只能設置為1 8 "FailCount": "${FailCount}|3",//處理失敗流程重試次數，如果出現異常，會進行重試 9 "prefetchCount": "${PrefetchCount}|0",//設置均勻分配消費者消息的個數 10 "BrokerName": "${BrokerName}|surging_demo",//exchange 名稱 11 "Port": "${EventBusPort}|5672"//埠 12 }
生成綁定的隊列如下圖
通過rabbitmq管理工具，可以通過properties來查看重試次數count 等一些信息，如下圖所示
4、如何部署
surging 服務引擎構建鏡像部署在docker中，可以按照業務需求自定義化引擎，也可以從 docker hub中pull鏡像，可以按照如下流程從docker hub 拉取部署鏡像
如何pull鏡像
可以通過命令：
docker pull serviceengine/surging
可以指定具體的tag來拉取，比如需要拉取v0.9.0.2，執行以下命令
docker pull serviceengine/surging:v0.9.0.2
如何配置
1.鏡像可以用環境變數設置相關參數，而通過以下的預設配置文件知曉如何通過環境變數配置參數，配置的規則：${環境變數名}|預設值
{ "Surging": { "Ip": "${Surging_Server_IP}|0.0.0.0", "WatchInterval": 30, "Port": "${Surging_Server_Port}|99", "MappingIp": "${Mapping_ip}", "MappingPort": "${Mapping_Port}", "Token": "true", "MaxConcurrentRequests": 20, "ExecutionTimeoutInMilliseconds": 30000, "Protocol": "${Protocol}|None", //Http、Tcp、None "RootPath": "${RootPath}|D:\\userapp", "Ports": { "HttpPort": "${HttpPort}|280", "WSPort": "${WSPort}|96" }, "RequestCacheEnabled": false, "Packages": [ { "TypeName": "EnginePartModule", "Using": "${UseEngineParts}|DotNettyModule;NLogModule;MessagePackModule;ConsulModule;HttpProtocolModule;WSProtocolModule;EventBusRabbitMQModule;" } ] }, //如果引用多個同類型的組件，需要配置Packages，如果是自定義按需引用，無需配置Packages "Consul": { "ConnectionString": "${Register_Conn}|127.0.0.1:8500", // "127.0.0.1:8500", "SessionTimeout": "${Register_SessionTimeout}|50", "RoutePath": "${Register_RoutePath}", "ReloadOnChange": true }, "EventBus_Kafka": { "Servers": "${EventBusConnection}|localhost:9092", "MaxQueueBuffering": "${MaxQueueBuffering}|10", "MaxSocketBlocking": "${MaxSocketBlocking}|10", "EnableAutoCommit": "${EnableAutoCommit}|false", "LogConnectionClose": "${LogConnectionClose}|false", "OffsetReset": "${OffsetReset}|earliest", "GroupID": "${EventBusGroupID}|surgingdemo" }, "EventBus": { "EventBusConnection": "${EventBusConnection}|localhost", "EventBusUserName": "${EventBusUserName}|guest", "EventBusPassword": "${EventBusPassword}|guest", "VirtualHost": "${VirtualHost}|/", "MessageTTL": "${MessageTTL}|30000", "RetryCount": "${RetryCount}|1", "FailCount": "${FailCount}|3", "BrokerName": "${BrokerName}|surging_demo", "Port": "${EventBusPort}|5672" }, "Zookeeper": { "ConnectionString": "${Zookeeper_ConnectionString}|127.0.0.1:2181", "SessionTimeout": 50, "ReloadOnChange": true }, "Logging": { "Debug": { "LogLevel": { "Default": "Information" } }, "Console": { "IncludeScopes": true, "LogLevel": { "Default": "${LogLevel}|Debug" } }, "LogLevel": { "Default": "${LogLevel}|Debug" } } }
2.可以通過設置環境變數surgingpath和cachepath來指定自定義文件配置，比如，掛載/home/fanly 目錄，通過以下命令參數 -v /home/fanly:/home/fanly 來設定，再通過設置以下命令參數用來設定自定義文件配置
--env surgingpath=/home/fanly/configs/surgingSettings.json
--env cachepath=/home/fanly/configs/cacheSettings.json
如何啟動內置引擎組件
引擎可以載入多個同一類型的引擎組件，可以通過以下配置啟用哪一種引擎組件,如果是自定義的服務引擎，不需要配置以下配置，只需要按照需求引用組件
"Packages": [ { "TypeName": "EnginePartModule", "Using": "${UseEngineParts}|DotNettyModule;NLogModule;MessagePackModule;ConsulModule;HttpProtocolModule;WSProtocolModule;EventBusRabbitMQModule;" } ]
如何啟動引擎
比如 pull 的鏡像是serviceengine/surging:v0.9.0.2 ，可以按照以下命令進行啟動
docker run --name surging --env surgingpath=/home/fanly/configs/surgingSettings.json --env cachepath=/home/fanly/configs/cacheSettings.json -v /home/fanly:/home/fanly serviceengine/surging:v0.9.0.2
7.總結
如有問題請到這裡提問 ，可以加入surging互相交流QQ群：542283494，引擎組件擴展溝通群：615562965
