【.NET Core專案實戰-統一認證平臺】第十五章閘道器篇-使用二級快取提升效能

Redis .NETCore · 發表 2019-03-05 17:47:00

摘要：【.NET Core專案實戰-統一認證平臺】開篇及目錄索引一、背景首先說聲抱歉，可能是因為假期綜合症（其實就是因為懶哈）的原因，已經很長時間沒更新部落格了，現在也調整的差不多了，準備還是以每週1-2篇的進度來更新部落格，並完成本專案所有功能。言歸正傳，本重構專案是在...

【.NET Core專案實戰-統一認證平臺】開篇及目錄索引

一、背景

首先說聲抱歉，可能是因為假期綜合症（其實就是因為懶哈）的原因，已經很長時間沒更新部落格了，現在也調整的差不多了，準備還是以每週1-2篇的進度來更新部落格，並完成本專案所有功能。

言歸正傳，本重構專案是在我根據實際需求重構，由於還未完全寫完，所以也沒進行壓測，在2月份時，張善友老師給我留言說經過壓測發現我重構的 Ocelot 閘道器功能效能較差，其中根本原因就是快取模組，由於重構專案的快取強依賴 Redis 快取，造成效能瓶頸，發現問題後，我也第一時間進行測試，效能影響很大， 經過跟張老師請教 ，可以使用二級快取來解決效能問題，首先感謝張老師關注並指點迷津，於是就有了這篇文章，如何把現有快取改成二級快取並使用。

二、改造思路

為了解決 redis 的強依賴性，首先需要把快取資料儲存到本地，所有請求都優先從本地提取，如果提取不到再從 redis 提取，如果 redis 無資料，在從資料庫中提取。提取流程如下：

MemoryCache > Redis > db

此種方式減少提取快取的網路開銷，也合理利用了分散式快取，並最終減少資料庫的訪問開銷。但是使用此種方案也面臨了一個問題是如何保證叢集環境時每個機器本地快取資料的一致性，這時我們會想到redis的釋出、訂閱特性，在資料發生變動時更新redis資料併發布快取更新通知，由每個叢集機器訂閱變更事件，然後處理本地快取記錄，最終達到叢集快取的快取一致性。

但是此方式對於快取變更非常頻繁的業務不適用，比如限流策略（準備還是使用分散式redis快取實現），但是可以擴充套件配置單機限流時使用本地快取實現，如果誰有更好的實現方式，也麻煩告知下叢集環境下限流的實現，不勝感激。

三、改造程式碼

首先需要分析下目前改造後的Ocelot閘道器在哪些業務中使用的快取，然後把使用本地快取的的業務重構，增加提取資料流程，最後提供閘道器外部快取初始化介面，便於與業務系統進行整合。

1.重寫快取方法

找到問題的原因後，就可以重寫快取方法，增加二級快取支援，預設使用本地的快取，新建 CzarMemoryCache 類，來實現 IOcelotCache<T> 方法，實現程式碼如下。

using Czar.Gateway.Configuration;
using Czar.Gateway.RateLimit;
using Microsoft.Extensions.Caching.Memory;
using Ocelot.Cache;
using System;

namespace Czar.Gateway.Cache
{
/// <summary>
/// 金焰的世界
/// 2019-03-03
/// 使用二級快取解決叢集環境問題
/// </summary>
public class CzarMemoryCache<T> : IOcelotCache<T>
{
private readonly CzarOcelotConfiguration _options;
private readonly IMemoryCache _cache;
public CzarMemoryCache(CzarOcelotConfiguration options,IMemoryCache cache)
{
_options = options;
_cache = cache;
}
public void Add(string key, T value, TimeSpan ttl, string region)
{
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix,region, key);
if (_options.ClusterEnvironment)
{
var msg = value.ToJson();
if (typeof(T) == typeof(CachedResponse))
{//帶過期時間的快取
_cache.Set(key, value, ttl); //新增本地快取
RedisHelper.Set(key, msg); //加入redis快取
RedisHelper.Publish(key, msg); //釋出
}
else if (typeof(T) == typeof(CzarClientRateLimitCounter?))
{//限流快取，直接使用redis
RedisHelper.Set(key, value, (int)ttl.TotalSeconds);
}
else
{//正常快取,釋出
_cache.Set(key, value, ttl); //新增本地快取
RedisHelper.Set(key, msg); //加入redis快取
RedisHelper.Publish(key, msg); //釋出
}
}
else
{
_cache.Set(key, value, ttl); //新增本地快取
}
}

public void AddAndDelete(string key, T value, TimeSpan ttl, string region)
{
Add(key, value, ttl, region);
}

public void ClearRegion(string region)
{
if (_options.ClusterEnvironment)
{
var keys = RedisHelper.Keys(region + "*");
RedisHelper.Del(keys);
foreach (var key in keys)
{
RedisHelper.Publish(key, ""); //釋出key值為空，處理時刪除即可。
}
}
else
{
_cache.Remove(region);
}
}

public T Get(string key, string region)
{
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, key);
if(region== CzarCacheRegion.CzarClientRateLimitCounterRegion&& _options.ClusterEnvironment)
{//限流且開啟了叢集支援，預設從redis取
return RedisHelper.Get<T>(key);
}
var result = _cache.Get<T>(key);
if (result == null&& _options.ClusterEnvironment)
{
result= RedisHelper.Get<T>(key);
if (result != null)
{
if (typeof(T) == typeof(CachedResponse))
{//檢視redis過期時間
var second = RedisHelper.Ttl(key);
if (second > 0)
{
_cache.Set(key, result, TimeSpan.FromSeconds(second));
}
}
else
{
_cache.Set(key, result, TimeSpan.FromSeconds(_options.CzarCacheTime));
}
}
}
return result;
}
}
}

上面就段程式碼實現了本地快取和 Redis 快取的支援，優先從本地提取，如果在叢集環境使用，增加 redis 快取支援，但是此種方式不適用快取變更非常頻繁場景，比如客戶端限流的實現，所以在程式碼中把客戶端限流的快取直接使用 redis 快取實現。

2.注入實現和訂閱

有了實現程式碼後，發現還缺少新增快取注入和配置資訊修改。首先需要修改配置檔案來滿足是否開啟叢集判斷，然後需要實現 redis 的不同部署方式能夠通過配置檔案配置進行管理，避免硬編碼導致的不可用問題。

配置檔案 CzarOcelotConfiguration.cs 修改程式碼如下：

namespace Czar.Gateway.Configuration
{
/// <summary>
/// 金焰的世界
/// 2018-11-11
/// 自定義配置資訊
/// </summary>
public class CzarOcelotConfiguration
{
/// <summary>
/// 資料庫連線字串，使用不同資料庫時自行修改,預設實現了SQLSERVER
/// </summary>
public string DbConnectionStrings { get; set; }

/// <summary>
/// 金焰的世界
/// 2018-11-12
/// 是否啟用定時器，預設不啟動
/// </summary>
public bool EnableTimer { get; set; } = false;

/// <summary>
/// 金焰的世界
/// 2018-11.12
/// 定時器週期，單位（毫秒），預設30分總自動更新一次
/// </summary>
public int TimerDelay { get; set; } = 30 * 60 * 1000;

/// <summary>
/// 金焰的世界
/// 2018-11-14
/// Redis連線字串
/// </summary>
public string RedisConnectionString { get; set; }

/// <summary>
/// 金焰的世界
/// 2019-03-03
/// 配置哨兵或分割槽時使用
/// </summary>
public string[] RedisSentinelOrPartitionConStr { get; set; }

/// <summary>
/// 金焰的世界
/// 2019-03-03
/// Redis部署方式，預設使用普通方式
/// </summary>
public RedisStoreMode RedisStoreMode { get; set; } = RedisStoreMode.Normal;

/// <summary>
/// 金焰的計界
/// 2019-03-03
/// 做叢集快取同步時使用，會訂閱所有正則匹配的事件
/// </summary>
public string RedisOcelotKeyPrefix { get; set; } = "CzarOcelot";

/// <summary>
/// 金焰的世界
/// 2019-03-03
/// 是否啟用叢集環境，如果非叢集環境直接本地快取+資料庫即可
/// </summary>
public bool ClusterEnvironment { get; set; } = false;

/// <summary>
/// 金焰的世界
/// 2018-11-15
/// 是否啟用客戶端授權,預設不開啟
/// </summary>
public bool ClientAuthorization { get; set; } = false;

/// <summary>
/// 金焰的世界
/// 2018-11-15
/// 伺服器快取時間，預設30分鐘
/// </summary>
public int CzarCacheTime { get; set; } = 1800;
/// <summary>
/// 金焰的世界
/// 2018-11-15
/// 客戶端標識，預設 client_id
/// </summary>
public string ClientKey { get; set; } = "client_id";

/// <summary>
/// 金焰的世界
/// 2018-11-18
/// 是否開啟自定義限流，預設不開啟
/// </summary>
public bool ClientRateLimit { get; set; } = false;
}
}

在配置檔案中修改了 redis 相關配置，支援使用 redis 的普通模式、叢集模式、哨兵模式、分割槽模式，配置方式可參考 csrediscore 開源專案。

然後修改 ServiceCollectionExtensions.cs 程式碼，注入相關實現和 redis 客戶端。

builder.Services.AddMemoryCache(); //新增本地快取
#region 啟動Redis快取，並支援普通模式 官方叢集模式哨兵模式 分割槽模式
if (options.ClusterEnvironment)
{
//預設使用普通模式
var csredis = new CSRedis.CSRedisClient(options.RedisConnectionString);
switch (options.RedisStoreMode)
{
case RedisStoreMode.Partition:
var NodesIndex = options.RedisSentinelOrPartitionConStr;
Func<string, string> nodeRule = null;
csredis = new CSRedis.CSRedisClient(nodeRule, options.RedisSentinelOrPartitionConStr);
break;
case RedisStoreMode.Sentinel:
csredis = new CSRedis.CSRedisClient(options.RedisConnectionString, options.RedisSentinelOrPartitionConStr);
break;
}
//初始化 RedisHelper
RedisHelper.Initialization(csredis);
}
#endregion
builder.Services.AddSingleton<IOcelotCache<FileConfiguration>, CzarMemoryCache<FileConfiguration>>();
builder.Services.AddSingleton<IOcelotCache<InternalConfiguration>, CzarMemoryCache<InternalConfiguration>>();
builder.Services.AddSingleton<IOcelotCache<CachedResponse>, CzarMemoryCache<CachedResponse>>();
builder.Services.AddSingleton<IInternalConfigurationRepository, RedisInternalConfigurationRepository>();
builder.Services.AddSingleton<IOcelotCache<ClientRoleModel>, CzarMemoryCache<ClientRoleModel>>();
builder.Services.AddSingleton<IOcelotCache<RateLimitRuleModel>, CzarMemoryCache<RateLimitRuleModel>>();
builder.Services.AddSingleton<IOcelotCache<RemoteInvokeMessage>, CzarMemoryCache<RemoteInvokeMessage>>();
builder.Services.AddSingleton<IOcelotCache<CzarClientRateLimitCounter?>, CzarMemoryCache<CzarClientRateLimitCounter?>>();

現在需要實現 redis 訂閱來更新本地的快取資訊，在專案啟動時判斷是否開啟叢集模式，如果開啟就啟動訂閱，實現程式碼如下：

public static async Task<IApplicationBuilder> UseCzarOcelot(this IApplicationBuilder builder, OcelotPipelineConfiguration pipelineConfiguration)
{
//重寫建立配置方法
var configuration = await CreateConfiguration(builder);
ConfigureDiagnosticListener(builder);
CacheChangeListener(builder);
return CreateOcelotPipeline(builder, pipelineConfiguration);
}
/// <summary>
/// 金焰的世界
/// 2019-03-03
/// 新增快取資料變更訂閱
/// </summary>
/// <param name="builder"></param>
/// <returns></returns>
private static void CacheChangeListener(IApplicationBuilder builder)
{
var config= builder.ApplicationServices.GetService<CzarOcelotConfiguration>();
var _cache= builder.ApplicationServices.GetService<IMemoryCache>();
if (config.ClusterEnvironment)
{
//訂閱滿足條件的所有事件
RedisHelper.PSubscribe(new[] { config.RedisOcelotKeyPrefix + "*" }, message =>
{
var key = message.Channel;
_cache.Remove(key); //直接移除，如果有請求從redis裡取
//或者直接判斷本地快取是否存在，如果存在更新，可自行實現。
});
}
}

使用的是從配置檔案提取的正則匹配的所有 KEY 都進行訂閱，由於本地快取增加了定時過期策略，所以為了實現方便，當發現 redis 資料發生變化，所有訂閱端直接移除本地快取即可，如果有新的請求直接從 redis 取，然後再次快取，防止叢集客戶端快取資訊不一致。

為了區分不同的快取實體，便於在原始資料傳送變更時進行更新，定義 CzarCacheRegion 類。

namespace Czar.Gateway.Configuration
{
/// <summary>
/// 快取所屬區域
/// </summary>
public class CzarCacheRegion
{
/// <summary>
/// 授權
/// </summary>
public const string AuthenticationRegion = "CacheClientAuthentication";

/// <summary>
/// 路由配置
/// </summary>
public const string FileConfigurationRegion = "CacheFileConfiguration";

/// <summary>
/// 內部配置
/// </summary>
public const string InternalConfigurationRegion = "CacheInternalConfiguration";

/// <summary>
/// 客戶端許可權
/// </summary>
public const string ClientRoleModelRegion = "CacheClientRoleModel";

/// <summary>
/// 限流規則
/// </summary>
public const string RateLimitRuleModelRegion = "CacheRateLimitRuleModel";

/// <summary>
/// Rpc遠端呼叫
/// </summary>
public const string RemoteInvokeMessageRegion = "CacheRemoteInvokeMessage";

/// <summary>
/// 客戶端限流
/// </summary>
public const string CzarClientRateLimitCounterRegion = "CacheCzarClientRateLimitCounter";
}
}

現在只需要修改快取的 region 為定義的值即可，唯一需要改動的程式碼就是把之前寫死的程式碼改成如下程式碼即可。

var enablePrefix = CzarCacheRegion.AuthenticationRegion;

3.開發快取變更介面

現在整個二級快取基本完成，但是還遇到一個問題就是外部如何根據資料庫變更資料時來修改快取資料，這時就需要提供外部修改 api 來實現。

新增 CzarCacheController.cs 對外部提供快取更新相關介面,詳細程式碼如下：

using Czar.Gateway.Authentication;
using Czar.Gateway.Configuration;
using Czar.Gateway.RateLimit;
using Czar.Gateway.Rpc;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Caching.Memory;
using Ocelot.Configuration;
using Ocelot.Configuration.Creator;
using Ocelot.Configuration.Repository;
using System;
using System.Threading.Tasks;

namespace Czar.Gateway.Cache
{
/// <summary>
/// 提供外部快取處理介面
/// </summary>
[Authorize]
[Route("CzarCache")]
public class CzarCacheController : Controller
{
private readonly CzarOcelotConfiguration _options;
private readonly IClientAuthenticationRepository _clientAuthenticationRepository;
private IFileConfigurationRepository _fileConfigurationRepository;
private IInternalConfigurationCreator _internalConfigurationCreator;
private readonly IClientRateLimitRepository _clientRateLimitRepository;
private readonly IRpcRepository _rpcRepository;
private readonly IMemoryCache _cache;
public CzarCacheController(IClientAuthenticationRepository clientAuthenticationRepository, CzarOcelotConfiguration options,
IFileConfigurationRepository fileConfigurationRepository,
IInternalConfigurationCreator internalConfigurationCreator,
IClientRateLimitRepository clientRateLimitRepository,
IRpcRepository rpcRepository,
IMemoryCache cache)
{
_clientAuthenticationRepository = clientAuthenticationRepository;
_options = options;
_fileConfigurationRepository = fileConfigurationRepository;
_internalConfigurationCreator = internalConfigurationCreator;
_clientRateLimitRepository = clientRateLimitRepository;
_rpcRepository = rpcRepository;
_cache = cache;
}

/// <summary>
/// 更新客戶端地址訪問授權介面
/// </summary>
/// <param name="clientid">客戶端ID</param>
/// <param name="path">請求模板</param>
/// <returns></returns>
[HttpPost]
[Route("ClientRule")]
public async Task UpdateClientRuleCache(string clientid, string path)
{
var region = CzarCacheRegion.AuthenticationRegion;
var key = CzarOcelotHelper.ComputeCounterKey(region, clientid, "", path);
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, key);
var result = await _clientAuthenticationRepository.ClientAuthenticationAsync(clientid, path);
var data = new ClientRoleModel() { CacheTime = DateTime.Now, Role = result };
if (_options.ClusterEnvironment)
{
RedisHelper.Set(key, data); //加入redis快取
RedisHelper.Publish(key, data.ToJson()); //釋出事件
}
else
{
_cache.Remove(key);
}
}

/// <summary>
/// 更新閘道器配置路由資訊
/// </summary>
/// <returns></returns>
[HttpPost]
[Route("InternalConfiguration")]
public async Task UpdateInternalConfigurationCache()
{
var key = CzarCacheRegion.InternalConfigurationRegion;
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, "", key);
var fileconfig = await _fileConfigurationRepository.Get();
var internalConfig = await _internalConfigurationCreator.Create(fileconfig.Data);
var config = (InternalConfiguration)internalConfig.Data;
if (_options.ClusterEnvironment)
{
RedisHelper.Set(key, config); //加入redis快取
RedisHelper.Publish(key, config.ToJson()); //釋出事件
}
else
{
_cache.Remove(key);
}
}

/// <summary>
/// 刪除路由配合的快取資訊
/// </summary>
/// <param name="region">區域</param>
/// <param name="downurl">下端路由</param>
/// <returns></returns>
[HttpPost]
[Route("Response")]
public async Task DeleteResponseCache(string region,string downurl)
{
var key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, downurl);
if (_options.ClusterEnvironment)
{
await RedisHelper.DelAsync(key);
RedisHelper.Publish(key, "");//釋出時間
}
else
{
_cache.Remove(key);
}
}

/// <summary>
/// 更新客戶端限流規則快取
/// </summary>
/// <param name="clientid">客戶端ID</param>
/// <param name="path">路由模板</param>
/// <returns></returns>
[HttpPost]
[Route("RateLimitRule")]
public async Task UpdateRateLimitRuleCache(string clientid, string path)
{
var region = CzarCacheRegion.RateLimitRuleModelRegion;
var key = clientid + path;
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, key);
var result = await _clientRateLimitRepository.CheckClientRateLimitAsync(clientid, path);
var data = new RateLimitRuleModel() { RateLimit = result.RateLimit, rateLimitOptions = result.rateLimitOptions };
if (_options.ClusterEnvironment)
{
RedisHelper.Set(key, data); //加入redis快取
RedisHelper.Publish(key, data.ToJson()); //釋出事件
}
else
{
_cache.Remove(key);
}
}

/// <summary>
/// 更新客戶端是否開啟限流快取
/// </summary>
/// <param name="path"></param>
/// <returns></returns>
[HttpPost]
[Route("ClientRole")]
public async Task UpdateClientRoleCache(string path)
{
var region = CzarCacheRegion.ClientRoleModelRegion;
var key = path;
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, key);
var result = await _clientRateLimitRepository.CheckReRouteRuleAsync(path);
var data = new ClientRoleModel() { CacheTime = DateTime.Now, Role = result };
if (_options.ClusterEnvironment)
{
RedisHelper.Set(key, data); //加入redis快取
RedisHelper.Publish(key, data.ToJson()); //釋出事件
}
else
{
_cache.Remove(key);
}
}

/// <summary>
/// 更新呢客戶端路由白名單快取
/// </summary>
/// <param name="clientid"></param>
/// <param name="path"></param>
/// <returns></returns>
[HttpPost]
[Route("ClientReRouteWhiteList")]
public async Task UpdateClientReRouteWhiteListCache(string clientid, string path)
{
var region = CzarCacheRegion.ClientReRouteWhiteListRegion;
var key = clientid + path;
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, key);
var result = await _clientRateLimitRepository.CheckClientReRouteWhiteListAsync(clientid, path);
var data = new ClientRoleModel() { CacheTime = DateTime.Now, Role = result };
if (_options.ClusterEnvironment)
{
RedisHelper.Set(key, data); //加入redis快取
RedisHelper.Publish(key, data.ToJson()); //釋出事件
}
else
{
_cache.Remove(key);
}
}

[HttpPost]
[Route("Rpc")]
public async Task UpdateRpcCache(string UpUrl)
{
var region = CzarCacheRegion.RemoteInvokeMessageRegion;
var key = UpUrl;
key = CzarOcelotHelper.GetKey(_options.RedisOcelotKeyPrefix, region, key);
var result = await _rpcRepository.GetRemoteMethodAsync(UpUrl);
if (_options.ClusterEnvironment)
{
RedisHelper.Set(key, result); //加入redis快取
RedisHelper.Publish(key, result.ToJson()); //釋出事件
}
else
{
_cache.Remove(key);
}
}
}
}

現在基本實現整個快取的更新策略，只要配合後臺管理介面，在相關快取原始資料傳送變更時，呼叫對應介面即可完成 redis 快取的更新，並自動通知叢集的所有本機清理快取等待重新獲取。

介面的呼叫方式參考之前我寫的配置資訊介面變更那篇即可。

四、效能測試

完成了改造後，我們拿改造前閘道器、改造後閘道器、原始Ocelot、直接呼叫API四個環境分別測試效能指標，由於測試環境有效，我直接使用本機環境，然後是 Apache ab測試工具 測試下相關效能（本測試不一定準確，只作為參考指標），測試的方式是使用100個併發請求10000次，測試結果分別如下。

1、改造前閘道器效能

2、改造後閘道器效能

3、Ocelot預設閘道器效能

4、直接呼叫API效能

本測試僅供參考，因為由於閘道器和服務端都在本機環境部署，所以使用閘道器和不使用閘道器效能差別非常小，如果分開部署可能性別差別會明顯寫，這不是本篇討論的重點。

從測試中可以看到，重構的閘道器改造前和改造後效能有2倍多的提升，且與原生的Ocelot效能非常接近。

五、總結

本篇主要講解了如何使用 redis 的釋出訂閱來實現二級快取功能，並提供了快取的更新相關介面供外部程式呼叫，避免出現叢集環境下無法更新快取資料導致提取資料不一致情況，但是針對每個客戶端獨立限流這塊叢集環境目前還是採用的 redis 的方式未使用本地快取，如果有寫的不對或有更好方式的，也希望多提寶貴意見。

本篇相關原始碼地址： https://github.com/jinyancao/czar.gateway

【.NET Core專案實戰-統一認證平臺】第十五章 閘道器篇-使用二級快取提升效能