第 11 章 開發(fā)實(shí)時應(yīng)用和服務(wù)

在本章，我們將討論“實(shí)時”的準(zhǔn)確含義，以及在大部分消費(fèi)者看來應(yīng)該屬于這一范疇的應(yīng)用類型

接著，我們將探討 WebSocket，并分析為什么傳統(tǒng)的 WebSocket 與云環(huán)境完全不相適應(yīng)，最后我們將構(gòu)建一個實(shí)時應(yīng)用的示例，用于展示向一個事件溯源系統(tǒng)添加實(shí)時消息的強(qiáng)大功能

實(shí)時應(yīng)用的定義

我認(rèn)為，實(shí)時系統(tǒng)的定義可以稍微寬泛一點(diǎn)，只要是事件的接收與處理過程之間只有少許延遲，或者完全沒有延遲都可以認(rèn)為是實(shí)時系統(tǒng)

下面是真正的實(shí)時系統(tǒng)中區(qū)分出非實(shí)時系統(tǒng)的幾個特點(diǎn)：

• 應(yīng)用收集輸入數(shù)據(jù)后，在生成輸出前，有明顯的等待

• 應(yīng)用只按照固定間隔或者基于某種按計劃或隨機(jī)觸發(fā)的外部信號生成輸出

實(shí)時系統(tǒng)有一個真正常見的跡象和特征，即當(dāng)相關(guān)方關(guān)注的事件發(fā)生時，它們會收到推送通知，而不是由相關(guān)方以掛起等待或者間隔查詢的方式來檢查新狀態(tài)

云環(huán)境中的 WebSocket

WebSocket 協(xié)議

WebSocket 協(xié)議始于 2008 年，它定義了瀏覽器和服務(wù)器之間建立持久的雙向 Socket 連接的標(biāo)準(zhǔn)

這讓服務(wù)器向運(yùn)行于瀏覽器中的 Web 應(yīng)用發(fā)送數(shù)據(jù)稱為可能，期間不需要由 Web 應(yīng)用執(zhí)行“輪詢”

在底層實(shí)現(xiàn)中，瀏覽器向服務(wù)器請求連接進(jìn)行升級

握手完成后，瀏覽器和服務(wù)器將切換為單獨(dú)的二進(jìn)制 TCP 連接，以實(shí)現(xiàn)雙向通信

部署模式

假如所有服務(wù)器都運(yùn)行在亞馬遜云的彈性計算服務(wù)環(huán)境中

當(dāng)虛擬機(jī)被托管在云基礎(chǔ)設(shè)施中時，它們就可能隨時被搬移、銷毀并重建

這原本是一件好事，旨在讓應(yīng)用近乎不受限制地伸縮

不過，這也意味著這種“實(shí)時” WebSocket 連接可能被切斷或者嚴(yán)重延遲，并在不知不覺中失去響應(yīng)

此處的解決方案通常是將對 WebSocket 的使用獨(dú)立出去--把管理 WebSocket 連接和數(shù)據(jù)傳輸工作轉(zhuǎn)移到應(yīng)用的代碼之外的位置

簡單地說，相比于在自己的應(yīng)用中管理 WebSocket，我們應(yīng)該選用一種基于云的消息服務(wù)，讓更專業(yè)的人來完成這項(xiàng)工作

使用云消息服務(wù)

我們的應(yīng)用需要擁有實(shí)時通信的能力

我們希望微服務(wù)能夠向客戶端推送數(shù)據(jù)，但客戶端無法建立到微服務(wù)的持續(xù) TCP 連接

我們還希望能夠使用相同類似的消息機(jī)制向后端服務(wù)發(fā)送消息

為讓微服務(wù)遵循云原生特性、保留可伸縮的能力，并在云環(huán)境中自由地搬移，我們需要挑選一種消息服務(wù)，把一定的實(shí)時通信能力提取到進(jìn)程之外

下面列舉一些廠商，他們提供的云消息服務(wù)有的是獨(dú)立產(chǎn)品，有的則是大型服務(wù)套件中的一部分：

• Apigee （API 網(wǎng)關(guān)與實(shí)時消息通信）

• PubNub （實(shí)時消息通信與活躍度監(jiān)控）

• Pusher（實(shí)時消息通信活躍度監(jiān)控）

• Kaazing（實(shí)時消息通信）

• Mashery（API 網(wǎng)關(guān)與實(shí)時消息通信）

• Google （Google 云消息通信）

• ASP.NET SinglR (Azure 托管的實(shí)時消息通信服務(wù))

• Amazon （簡單通知服務(wù)）

無論選擇哪種機(jī)制，我們都應(yīng)該投入一定的時間讓代碼與具體的消息服務(wù)相隔離，從而在更換服務(wù)商時，不至于產(chǎn)生太大的影響

開發(fā)位置接近監(jiān)控服務(wù)

現(xiàn)在，我們要做的就是開發(fā)一個每當(dāng)后端系統(tǒng)檢測到接近事件時，就能夠?qū)崟r更新的監(jiān)視器

我們可以生成一張地圖，在上面繪出兩個團(tuán)隊(duì)成員的位置，當(dāng)系統(tǒng)檢測到他們相互接近時，就讓他們的頭像跳動，或者生成一個動畫

這些團(tuán)隊(duì)成員的移動設(shè)備可能還會在同一時刻收到通知

創(chuàng)建接近監(jiān)控服務(wù)

我們的示例監(jiān)控服務(wù)將包含一系列不同的組件

首先，我們需要消費(fèi)由第 6 章編寫的服務(wù)生成并放入隊(duì)列的 ProximityDetectedEvent 事件

此后，我們要提取事件中的原始信息，調(diào)用團(tuán)隊(duì)服務(wù)以獲取可供用戶讀取識別的信息

獲取這些補(bǔ)充信息后，最后要在實(shí)時消息系統(tǒng)上發(fā)出一條消息

GitHub鏈接：https://github.com/microservices-aspnetcore/es-proximitymonitor

以下是我們接近監(jiān)控服務(wù)背后的上層協(xié)調(diào)邏輯

using System; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using StatlerWaldorfCorp.ProximityMonitor.Queues; using StatlerWaldorfCorp.ProximityMonitor.Realtime; using StatlerWaldorfCorp.ProximityMonitor.TeamService;namespace StatlerWaldorfCorp.ProximityMonitor.Events {public class ProximityDetectedEventProcessor : IEventProcessor{private ILogger logger;private IRealtimePublisher publisher;private IEventSubscriber subscriber;private PubnubOptions pubnubOptions;public ProximityDetectedEventProcessor(ILogger<ProximityDetectedEventProcessor> logger,IRealtimePublisher publisher,IEventSubscriber subscriber,ITeamServiceClient teamClient,IOptions<PubnubOptions> pubnubOptions){this.logger = logger;this.pubnubOptions = pubnubOptions.Value;this.publisher = publisher;this.subscriber = subscriber;logger.LogInformation("Created Proximity Event Processor.");subscriber.ProximityDetectedEventReceived += (pde) => {Team t = teamClient.GetTeam(pde.TeamID);Member sourceMember = teamClient.GetMember(pde.TeamID, pde.SourceMemberID);Member targetMember = teamClient.GetMember(pde.TeamID, pde.TargetMemberID);ProximityDetectedRealtimeEvent outEvent = new ProximityDetectedRealtimeEvent{TargetMemberID = pde.TargetMemberID,SourceMemberID = pde.SourceMemberID,DetectionTime = pde.DetectionTime,SourceMemberLocation = pde.SourceMemberLocation,TargetMemberLocation = pde.TargetMemberLocation,MemberDistance = pde.MemberDistance,TeamID = pde.TeamID,TeamName = t.Name,SourceMemberName = $"{sourceMember.FirstName} {sourceMember.LastName}",TargetMemberName = $"{targetMember.FirstName} {targetMember.LastName}"};publisher.Publish(this.pubnubOptions.ProximityEventChannel, outEvent.toJson());};}public void Start(){subscriber.Subscribe();}public void Stop(){subscriber.Unsubscribe();}} }

在這個代碼清單中，首先要注意的是從 DI 向構(gòu)造函數(shù)注入的一連串依賴：

• 日志記錄工具

• 實(shí)時事件發(fā)布器

• 事件訂閱器

• 團(tuán)隊(duì)服務(wù)客戶端

• PubNub 選項(xiàng)

創(chuàng)建實(shí)時事件發(fā)布器類實(shí)現(xiàn)類

using Microsoft.Extensions.Logging; using PubnubApi;namespace StatlerWaldorfCorp.ProximityMonitor.Realtime {public class PubnubRealtimePublisher : IRealtimePublisher{private ILogger logger;private Pubnub pubnubClient;public PubnubRealtimePublisher(ILogger<PubnubRealtimePublisher> logger,Pubnub pubnubClient){logger.LogInformation("Realtime Publisher (Pubnub) Created.");this.logger = logger;this.pubnubClient = pubnubClient;}public void Validate(){pubnubClient.Time().Async(new PNTimeResultExt((result, status) => {if (status.Error) {logger.LogError($"Unable to connect to Pubnub {status.ErrorData.Information}");throw status.ErrorData.Throwable;} else {logger.LogInformation("Pubnub connection established.");}}));}public void Publish(string channelName, string message){pubnubClient.Publish().Channel(channelName).Message(message).Async(new PNPublishResultExt((result, status) => {if (status.Error) {logger.LogError($"Failed to publish on channel {channelName}: {status.ErrorData.Information}");} else {logger.LogInformation($"Published message on channel {channelName}, {status.AffectedChannels.Count} affected channels, code: {status.StatusCode}");}}));}} }

注入實(shí)時通信類

在 Startup 類中配置 DI 來提供 PubNub 客戶端和其他相關(guān)類

using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using StatlerWaldorfCorp.ProximityMonitor.Queues; using StatlerWaldorfCorp.ProximityMonitor.Realtime; using RabbitMQ.Client.Events; using StatlerWaldorfCorp.ProximityMonitor.Events; using Microsoft.Extensions.Options; using RabbitMQ.Client; using StatlerWaldorfCorp.ProximityMonitor.TeamService;namespace StatlerWaldorfCorp.ProximityMonitor {public class Startup{public Startup(IHostingEnvironment env, ILoggerFactory loggerFactory){loggerFactory.AddConsole();loggerFactory.AddDebug();var builder = new ConfigurationBuilder().SetBasePath(env.ContentRootPath).AddJsonFile("appsettings.json", optional: false, reloadOnChange: false).AddEnvironmentVariables();Configuration = builder.Build();}public IConfigurationRoot Configuration { get; }public void ConfigureServices(IServiceCollection services){services.AddMvc();services.AddOptions();services.Configure<QueueOptions>(Configuration.GetSection("QueueOptions"));services.Configure<PubnubOptions>(Configuration.GetSection("PubnubOptions"));services.Configure<TeamServiceOptions>(Configuration.GetSection("teamservice"));services.Configure<AMQPOptions>(Configuration.GetSection("amqp"));services.AddTransient(typeof(IConnectionFactory), typeof(AMQPConnectionFactory));services.AddTransient(typeof(EventingBasicConsumer), typeof(RabbitMQEventingConsumer));services.AddSingleton(typeof(IEventSubscriber), typeof(RabbitMQEventSubscriber));services.AddSingleton(typeof(IEventProcessor), typeof(ProximityDetectedEventProcessor));services.AddTransient(typeof(ITeamServiceClient),typeof(HttpTeamServiceClient));services.AddRealtimeService();services.AddSingleton(typeof(IRealtimePublisher), typeof(PubnubRealtimePublisher));}// Singletons are lazy instantiation.. so if we don't ask for an instance during startup,// they'll never get used.public void Configure(IApplicationBuilder app,IHostingEnvironment env,ILoggerFactory loggerFactory,IEventProcessor eventProcessor,IOptions<PubnubOptions> pubnubOptions,IRealtimePublisher realtimePublisher){realtimePublisher.Validate();realtimePublisher.Publish(pubnubOptions.Value.StartupChannel, "{'hello': 'world'}");eventProcessor.Start();app.UseMvc();}} }

我們嘗試為類提供預(yù)先創(chuàng)建好的 PubNub API 實(shí)例

為整潔地實(shí)現(xiàn)這一功能，并繼續(xù)以注入方式獲取配置信息，包括 API 密鑰，我們需要向 DI 中注冊一個工廠

工廠類的職責(zé)是向外提供裝配完成的 PubNub 實(shí)例

using System; using Microsoft.Extensions.Options; using PubnubApi; using System.Linq; using Microsoft.Extensions.Logging;namespace StatlerWaldorfCorp.ProximityMonitor.Realtime {public class PubnubFactory{private PNConfiguration pnConfiguration;private ILogger logger;public PubnubFactory(IOptions<PubnubOptions> pubnubOptions,ILogger<PubnubFactory> logger){this.logger = logger;pnConfiguration = new PNConfiguration();pnConfiguration.PublishKey = pubnubOptions.Value.PublishKey;pnConfiguration.SubscribeKey = pubnubOptions.Value.SubscribeKey;pnConfiguration.Secure = false;logger.LogInformation($"Pubnub Factory using publish key {pnConfiguration.PublishKey}");}public Pubnub CreateInstance(){return new Pubnub(pnConfiguration);}} }

將工廠注冊到 DI 時使用的擴(kuò)展方法機(jī)制

using System; using Microsoft.Extensions.DependencyInjection; using PubnubApi;namespace StatlerWaldorfCorp.ProximityMonitor.Realtime {public static class RealtimeServiceCollectionExtensions{public static IServiceCollection AddRealtimeService(this IServiceCollection services){services.AddTransient<PubnubFactory>();return AddInternal(services, p => p.GetRequiredService<PubnubFactory>(), ServiceLifetime.Singleton);}private static IServiceCollection AddInternal(this IServiceCollection collection,Func<IServiceProvider, PubnubFactory> factoryProvider,ServiceLifetime lifetime){Func<IServiceProvider, object> factoryFunc = provider =>{var factory = factoryProvider(provider);return factory.CreateInstance();};var descriptor = new ServiceDescriptor(typeof(Pubnub), factoryFunc, lifetime);collection.Add(descriptor);return collection;}} }

上面代碼的關(guān)鍵功能是創(chuàng)建了一個 lambda 函數(shù)，接收 IServiceProvider 作為輸入，并返回一個對象作為輸出

它正是我們注冊工廠時向服務(wù)描述對象中傳入的工廠方法

匯總所有設(shè)計

要立即查看效果，從而確保一切工作正常，我們可模擬由第 6 章的服務(wù)輸出的信息

只需要手動向 proximitydetected 隊(duì)列中放入表示 ProximityDetectedEvent 對象的 JSON 字符串

在這個過程中，如果我們的監(jiān)控服務(wù)處于運(yùn)行之中、訂閱了隊(duì)列，而且團(tuán)隊(duì)服務(wù)處于運(yùn)行之中、擁有正確的數(shù)據(jù)，那么接近監(jiān)控服務(wù)將取出事件、補(bǔ)充必要的數(shù)據(jù)，并通過 PubNub 發(fā)送一個實(shí)時事件

利用 PubNub 調(diào)試控制臺，我們可以立即看到這一處理過程生成的輸出

為實(shí)時接近監(jiān)控服務(wù)創(chuàng)建界面

為簡化工作，同時掩蓋我缺乏藝術(shù)細(xì)胞的真相，我將用一個不包含圖形元素的簡單 HTML 頁面，它不需要托管在專門的 Web 服務(wù)器上

它實(shí)時地監(jiān)聽接近事件，并將攜帶的信息動態(tài)添加到新的 div 元素中

realtimetest.html

<html><head><title>RT page sample</title><script src="https://cdn.pubnub.com/sdk/javascript/pubnub.4.4.0.js"></script><script>var pubnub = new PubNub({subscribeKey: "yoursubkey",publishKey: "yourprivatekey",ssl: true});pubnub.addListener({message: function(m) {// handle messagevar channelName = m.channel; // The channel for which the message belongsvar channelGroup = m.subscription; // The channel group or wildcard subscription match (if exists)var pubTT = m.timetoken; // Publish timetokenvar msg = JSON.parse(m.message); // The Payloadconsole.log("New Message!!", msg);var newDiv = document.createElement('div')var newStr = "** (" + msg.TeamName + ") " + msg.SourceMemberName + " moved within " + msg.MemberDistance + "km of " + msg.TargetMemberName;newDiv.innerHTML = newStrvar oldDiv = document.getElementById('chatLog')oldDiv.appendChild(newDiv)},presence: function(p) {// handle presencevar action = p.action; // Can be join, leave, state-change or timeoutvar channelName = p.channel; // The channel for which the message belongsvar occupancy = p.occupancy; // No. of users connected with the channelvar state = p.state; // User Statevar channelGroup = p.subscription; // The channel group or wildcard subscription match (if exists)var publishTime = p.timestamp; // Publish timetokenvar timetoken = p.timetoken; // Current timetokenvar uuid = p.uuid; // UUIDs of users who are connected with the channel},status: function(s) {// handle status}});console.log("Subscribing..");pubnub.subscribe({channels: ['proximityevents']});</script></head><body><h1>Proximity Monitor</h1><p>Proximity Events listed below.</p><div id="chatLog"></div></body> </html>

值得指出的是，這個文件并不需要托管在服務(wù)器上

在任何瀏覽器中打開，其中的 JavaScript 都可以運(yùn)行

總結(jié)

以上是生活随笔為你收集整理的《ASP.NET Core 微服务实战》-- 读书笔记（第11章）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。