MicroService（Ocelot）

为了防止调用服务主动权在客户端，让客户端调网关，网关调服务

网关作用：

1,重试策略（Retry）：当服务调用失败时，Polly 可以自动进行重试，这有助于
处理那些可能因为暂时性问题导致的服务不可用情况。
2、断路器（Circuit Breaker）：当检测到服务连续不可用时，断路器策略会介入，
快速返回错误响应，避免对下游服务的持续请求，从而预防服务雪崩现象。(熔断
ocelot+polly)
polly:瞬态故障处理库
3、超时策略（Timeout）：为服务调用设置一个最大执行时间，超过这个时间的服
务调用将被认为失败，可以采取预设的应对措施。
4、舱壁隔离（Bulkhead Isolation）：通过限制对服务的并发调用数量，防止因
某个服务的问题影响到整个系统的稳定性。（ocelot 限流）
5、缓存策略（Cache）：提供一种机制，可以在服务调用的结果不变的情况下直接
使用缓存结果，减少不必要的服务调用。（ocelot 缓存）
6、降级策略（Fallback）：当服务调用失败时，可以提供一个备用的逻辑或者数
据作为响应，提高用户体验。(polly 服务降级)
7、策略组合（PolicyWrap) : Polly 针对不同

1下载nuget包

 <PackageReference Include="Ocelot" Version="23.1.0" />
 <PackageReference Include="Ocelot.Provider.Consul" Version="23.1.0" />
 <PackageReference Include="Ocelot.Provider.Polly" Version="23.1.0" />

2基本认识Ocelot.json文件

////***************************单地址*ַ********************************
//访问: http://localhost:9005/T9001/Users/All
{
  "Routes": [
    {
      "DownstreamPathTemplate": "/api/{url}", //服务地址,Url变量
      "DownstreamScheme": "http",
      "DownstreamHostAndPorts": [
        {
          "Host": "127.0.0.1",
          "Port": 9001 //服务端口
        } //http://127.0.0.1:9001/api/users/all
      ],
      "UpstreamPathTemplate": "/T9001/{url}", //网关地址: Url变量
      "UpstreamHttpMethod": [ "Get", "Post" ]
    }
  ]
}


  //*****************************多地址********************************
//访问: https://localhost:7253/T9001/Users/All
//访问: https://localhost:7253/T9002/Users/All
//访问: https://localhost:7253/T9003/Users/All
{
  "Routes": [
    {
      "DownstreamPathTemplate": "/api/{url}", //服务地址,Url变量
      "DownstreamScheme": "http",
      "DownstreamHostAndPorts": [
        {
          "Host": "127.0.0.1",
          "Port": 9001 //服务端口
        }
      ],
      "UpstreamPathTemplate": "/T9001/{url}", //网关地址: Url变量
      "UpstreamHttpMethod": [ "Get", "Post" ]
    },
    {
      "DownstreamPathTemplate": "/api/{url}", //服务地址,Url变量
      "DownstreamScheme": "http",
      "DownstreamHostAndPorts": [
        {
          "Host": "127.0.0.1",
          "Port": 9002 //服务端口
        }
      ],
      "UpstreamPathTemplate": "/T9002/{url}", //网关地址: Url变量
      "UpstreamHttpMethod": [ "Get", "Post" ]
    },
    {
      "DownstreamPathTemplate": "/api/{url}", //服务地址,Url变量
      "DownstreamScheme": "http",
      "DownstreamHostAndPorts": [
        {
          "Host": "127.0.0.1",
          "Port": 9003 //服务端口
        }
      ],
      "UpstreamPathTemplate": "/T9003/{url}", //网关地址: Url变量
      "UpstreamHttpMethod": [ "Get", "Post" ]
    }
  ]
}


  //*****************************单地址多实例负载均衡********************************
//可以负载均衡--但是不能动态伸缩服务, 所以一般不会使用他来做负载均衡, 需要和Consul做集成
{
  "Routes": [
    {
      "DownstreamPathTemplate": "/api/{url}", //服务地址Url变量
      "DownstreamScheme": "http",
      "DownstreamHostAndPorts": [
        {
          "Host": "127.0.0.1",
          "Port": 9001
        },
        {
          "Host": "127.0.0.1",
          "Port": 9002
        },
        {
          "Host": "127.0.0.1",
          "Port": 9003
        } //能负载均衡,但是不能动态伸缩, 需要结合Consul来完成
      ],
      "UpstreamPathTemplate": "/T/{url}", //网关地址: Url变量
      "UpstreamHttpMethod": [ "Get", "Post" ],
      "LoadBalancerOptions": {
        "Type": "RoundRobin" //轮询 // "LeastConnection" //最小连接数的服务器  "NoLoadBalance" //不负载均衡 //
      }
      //"LoadBalancerOptions": {
      //  "Type": "CookieStickySessions",
      //  "Key": "ASP.NET_SessionId",
      //  "Expiry": 1800000
      //}
    }
  ]
}

完整文件，包含ocelot+consul+Polly+缓存

3使用Ocelot

{
  "Routes": [ //这里注意一下版本（旧版本用ReRoutes）
    {
      "DownstreamPathTemplate": "/{controller}",
      "DownstreamScheme": "http", //下游方案
      "UpstreamPathTemplate": "/TestGate/{controller}",
      "UpstreamHttpMethod": [ "Get", "Post" ],
      "UseServiceDiscovery": true,//使用服务发现
      "ServiceName": "ConsulService", //请求服务名称
      "LoadBalancerOptions": {
        "Type": "RoundRobin"//LeastConnection最少连接数的服务器，NoLoadBalance不负债均衡CookieStickySessions会话粘滞
      },
      "FileCacheOptions": {
        "TtlSeconds": 10,
        "Region": "myRegion"
      },
      "RateLimitOptions": {//限流
        "ClientWhitelist": ["Admin"],//白名单,在其中的不限流,请求头加上ClientId:Admin,区分大小写
        "EnableRateLimiting": true,
        "Period": "1m",//1s,1h
        "PeriodTimespan": 5,//多少秒可以重试
        "Limit": 4//时间段允许的最大请求数量
      },
      "QoSOptions": {
        "ExceptionsAllowedBeforeBreaking": 3,//允许3次异常polly
        "DurationOfBreak": 5000, //ms，熔断时间
        "TimeoutValue": 2000//2s内不给响应，异常时间超过2s,熔断
      }
    }
  ],
  "GlobalConfiguration": {//服务发现中心
    "BaseUrl": "http://localhost:6001", //进行标头查找和替换以及某些管理配置
    "ServiceDiscoveryProvider": {
      "Scheme": "http",
      "Host": "127.0.0.1", //你的Consul的ip地址
      "Port": 8500, //你的Consul的端口
      "Type": "Consul" //类型
    },
    "RateLimitOptions": {//限流响应
      "QuotaExceededMessage": "Too Many Requests,please wait a moment",
      "HttpStatusCode": 503
    }
  }
}

using Ocelot.DependencyInjection;
using Ocelot.Middleware;
using Ocelot.Provider.Consul;
using Ocelot.Provider.Polly;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.

builder.Services.AddControllers();
// Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();
builder.Services.AddOcelot().AddConsul().AddPolly();//注册

builder.Configuration.AddJsonFile("ocelot.json", false, true);
var app = builder.Build();

// Configure the HTTP request pipeline.
if (app.Environment.IsDevelopment())
{
    app.UseSwagger();
    app.UseSwaggerUI();
}

app.UseHttpsRedirection();

app.UseAuthorization();

app.MapControllers();
app.UseOcelot().Wait();//使用Ocelot
app.Run();

4进阶

扩展自定义缓存

namespace LearnOcelot.OcelotExtend
{

    public class CustomCacheExtend : IOcelotCache<CachedResponse>
    {

        private ILogger<CustomCacheExtend> _logger = null;
        public CustomCacheExtend(ILogger<CustomCacheExtend> logger)
        {
            this._logger = logger;
        }
         
        private static Dictionary<string, CacheDataModel> CustomCacheExtendDictionary = new
          Dictionary<string, CacheDataModel>();

        /// <summary>
        /// 向缓存中去缓存数据--会调用这个方法
        /// </summary>
        /// <param name="key"></param>
        /// <param name="value"></param>
        /// <param name="ttl"></param>
        /// <param name="region"></param>
        public void Add(string key, CachedResponse value, TimeSpan ttl, string region)
        {
            this._logger.LogWarning($"This is {nameof(CustomCacheExtend)}.{nameof(Add)}");
            //CustomCacheExtendDictionary.Add(key, new CacheDataModel()
            //{
            //    CachedResponse = value,
            //    Region = region,
            //    Timeout = DateTime.Now.Add(ttl)
            //});
            CustomCacheExtendDictionary[key] = new CacheDataModel()
            {
                CachedResponse = value,
                Region = region,
                Timeout = DateTime.Now.Add(ttl)
            };
        }

        /// <summary>
        /// 添加缓存, 如果有历史缓存, 就覆盖
        /// </summary>
        /// <param name="key"></param>
        /// <param name="value"></param>
        /// <param name="ttl"></param>
        /// <param name="region"></param>
        public void AddAndDelete(string key, CachedResponse value, TimeSpan ttl, string region)
        {
            this._logger.LogWarning($"This is {nameof(CustomCacheExtend)}.{nameof(AddAndDelete)}");
            CustomCacheExtendDictionary[key] = new CacheDataModel()
            {
                CachedResponse = value,
                Region = region,
                Timeout = DateTime.Now.Add(ttl)
            };
        }

        /// <summary>
        /// 清除缓存的数据
        /// </summary>
        /// <param name="region"></param>
        public void ClearRegion(string region)
        {
            this._logger.LogWarning($"This is {nameof(CustomCacheExtend)}.{nameof(ClearRegion)}");

            var keyList = CustomCacheExtendDictionary.Where(kv => kv.Value.Region.Equals(region)).Select(kv => kv.Key);
            foreach (var key in keyList)
            {
                CustomCacheExtendDictionary.Remove(key);
            }
        }

        public CachedResponse Get(string key, string region)
        {
            this._logger.LogWarning($"This is {nameof(CustomCacheExtend)}.{nameof(Get)}");
            if (CustomCacheExtendDictionary.ContainsKey(key) && CustomCacheExtendDictionary[key] != null
                && CustomCacheExtendDictionary[key].Timeout > DateTime.Now
                && CustomCacheExtendDictionary[key].Region.Equals(region))
            {
                return CustomCacheExtendDictionary[key].CachedResponse;
            }
            else
                return null;
        }
    }


    public class CacheDataModel
    {
        public CachedResponse CachedResponse { get; set; }
        public DateTime Timeout { get; set; }
        public string Region { get; set; }
    }
}

Program里添加

builder.Services.AddSingleton<IOcelotCache<CachedResponse>, CustomCacheExtend>();

AddandDelete和ClearRegion需要你自己调用，用于清除脏数据

Add方法在Get方法返回null调用。

扩展自定义轮询

public class CustomPollingLoadBalancer : ILoadBalancer
{
    /// <summary>
    /// 通过注入的一段逻辑---就是在IOC注册的时候, 指定的策略的规则
    /// </summary>
    private readonly Func<Task<List<Service>>> _DownstreamServicesTaskList;

    /// <summary>
    /// 锁
    /// </summary>
    private readonly object CustomPollingLoadBalancer_Lock = new object();
    private int _lastIndex;

    //内部自己注册的ioc的
    public CustomPollingLoadBalancer(Func<Task<List<Service>>> services)
    {
        this._DownstreamServicesTaskList = services;
    }


    /// <summary>
    /// 决定负载均衡的策略
    /// </summary>
    /// <param name="httpContext"></param>
    /// <returns></returns>
    /// <exception cref="NotImplementedException"></exception>
    public async Task<Response<ServiceHostAndPort>> Lease(HttpContext httpContext)
    {

        ///downstreamServices: 服务列表
        var downstreamServices = await this._DownstreamServicesTaskList();
        lock (CustomPollingLoadBalancer_Lock)
        {
            Console.WriteLine($"This is {nameof(CustomPollingLoadBalancer)}.Lease");
            Console.WriteLine($"This is {httpContext.Request.Host.Value}");
            Console.WriteLine($"This is {string.Join(",", downstreamServices.Select(s => s.HostAndPort.DownstreamHost + s.HostAndPort.DownstreamPort))}");


            if (_lastIndex >= downstreamServices.Count)
            {
                _lastIndex = 0;
            } 

            var next = downstreamServices[_lastIndex];
            _lastIndex++;
            return new OkResponse<ServiceHostAndPort>(next.HostAndPort);
        }
    }

    public void Release(ServiceHostAndPort hostAndPort)
    {
       
    }
}

Program.cs里添加

#region IOC扩展负载均衡策略
Func<IServiceProvider, DownstreamRoute, IServiceDiscoveryProvider, CustomPollingLoadBalancer> loadBalancerFactoryFunc = (serviceProvider, Route, serviceDiscoveryProvider) => new CustomPollingLoadBalancer(serviceDiscoveryProvider.Get);
#endregion



builder.Services.AddOcelot()
    //注意: nuget引入: Ocelot.Provider.Consul
    .AddConsul() //Ocelot注册到IOC容器
                 //扩展 Ocelot-缓存
                 //Nuet: Ocelot.Cache.CacheManager
    .AddCacheManager(x =>
    {
        x.WithDictionaryHandle();//字典缓存
        //如果我自己来扩展下, 可能会使用redis  Mongo  memaryCache 
    })
    .AddCustomLoadBalancer<CustomPollingLoadBalancer>(loadBalancerFactoryFunc);  //配置生效自定义负载均衡策略

文章版权归作者所有，未经允许请勿转载。如内容涉嫌侵权，请在本页底部进入<联系我们>进行举报投诉!

THE END