.Net Core使用Consul+Ocelot搭建微服务项目

时代在变,技术也在更新迭代。从传统的单体应用架构到现在的分布式集群架构,在技术的学习上真的是一点都不能松懈。

网上关于微服务与Consul的话题太多了,我在这里不做过多描述。

其实就是在微服务中我们可以利用Consul可以实现服务的发现、治理、健康检查等...

用它先下载它：

https://www.consul.io/downloads.html

我此番在windows下操作,打开下载的Consul所在文件夹，输入 consul.exe agent -dev

Consul的默认启动端口为8500,如果能正常显示页面则启动成功。

 新建解决方案,建立一个.Net Core MVC的项目和一个.Net Core WebApi的项目。 安装NuGet包Consul

首先Api端服务实例启动时需到Consul中进行服务注册,Web Client直接与Consul进行连接,从Consul中拿到服务实例并配置策略及发送http请求等。

如图所示：

 Consul每隔一段时间就会调用一次注册的服务实例进行健康检查。

在Api项目中新建一个IConfiguration的扩展方法：

public static void ConsulExtend(this IConfiguration configuration)
{
    ConsulClient client = new ConsulClient(m =>
    {
        m.Address = new Uri("http://localhost:8500/");
        m.Datacenter = "dc1";
    });
    //启动的时候在consul中注册实例服务
    //在consul中注册的ip,port
    string ip = configuration["ip"];
    int port = int.Parse(configuration["port"]);
    int weight = string.IsNullOrWhiteSpace(configuration["weight"]) ? 1 : int.Parse(configuration["weight"]);
    client.Agent.ServiceRegister(new AgentServiceRegistration()
    {
        ID = "service" + Guid.NewGuid(),//唯一的
        Name = "MicroserviceAttempt",//组(服务)名称
        Address = ip,
        Port = port,//不同的端口=>不同的实例
        Tags = new string[] { weight.ToString() },//标签
        Check = new AgentServiceCheck()//服务健康检查
        {
            Interval = TimeSpan.FromSeconds(12),//间隔12s一次 检查
            HTTP = $"http://{ip}:{port}/Api/Health/Index",
            Timeout = TimeSpan.FromSeconds(5),//检测等待时间
            DeregisterCriticalServiceAfter = TimeSpan.FromSeconds(20)//失败后多久移除
        }
    });
    Console.WriteLine($"{ip}:{port}--weight:{weight}");
}

心跳检查的接口：

[ApiController]
[Route("api/[controller]/[action]")]
public class HealthController : Controller
{
    readonly IConfiguration _configuration;
    public HealthController(IConfiguration configuration)
    {
        _configuration = configuration;
    }
    [HttpGet]
    public IActionResult Index()
    {
        //心跳,consul会每隔几秒调一次
        Console.WriteLine($"{ _configuration["port"]} Invoke");
        return Ok();
    }
}

在Startup类中的Configure方法加入：

//启动时注册,且注册一次
this.Configuration.ConsulExtend();

将api项目启动(三个端口)

dotnet ServicesInstances.dll --urls="http://*:5726" --ip="127.0.0.1" --port=5726
dotnet ServicesInstances.dll --urls="http://*:5727" --ip="127.0.0.1" --port=5727
dotnet ServicesInstances.dll --urls="http://*:5728" --ip="127.0.0.1" --port=5728

接着是web端,新建控制器：

public class UserController : Controller
{readonly HttpSender _httpSender;
    public UserController(HttpSender httpSender)
    {
        _httpSender = httpSender;
    }
    //暂不考虑线程安全
    private static int index = 0;
    public async Task<IActionResult> Index()
    {
        #region nginx版 只知道nginx地址就行了
        //var str = await _httpSender.InvokeApi("http://localhost:8088/api/User/GetCustomerUser");
        #endregion

        #region consul
        //new一个consul实例
        ConsulClient client = new ConsulClient(m =>
        {
            new Uri("http://localhost:8500/");
            m.Datacenter = "dc1";
        });
        //与consul进行通信(连接),得到consul中所有的服务实例
        var response = client.Agent.Services().Result.Response;
        string url = "http://MicroserviceAttempt/api/User/GetCustomerUser";
        Uri uri = new Uri(url);
        string groupName = uri.Host;
        AgentService agentService = null;//服务实例
        var serviceDictionary = response.Where(m => m.Value.Service.Equals(groupName, StringComparison.OrdinalIgnoreCase)).ToArray();//找到的全部服务实例
        //{
        //    agentService = serviceDictionary[0].Value;
        //}
        {
            //轮询策略=>达到负载均衡的目的
            agentService = serviceDictionary[index++ % 3].Value;
        }
        {
            //平均策略（随机获取索引--相对平均）=>达到负载均衡的目的
            agentService = serviceDictionary[new Random(index++).Next(0, serviceDictionary.Length)].Value;
        }
        {
            //权重策略,给不同的实例分配不同的压力,注册时提供权重
            List<KeyValuePair<string, AgentService>> keyValuePairs = new List<KeyValuePair<string, AgentService>>();
            foreach (var item in keyValuePairs)
            {
                int count = int.Parse(item.Value.Tags?[0]);//在服务注册的时候给定权重数量
                for (int i = 0; i < count; i++)
                {
                    keyValuePairs.Add(item);
                }
            }
            agentService = keyValuePairs.ToArray()[new Random(index++).Next(0, keyValuePairs.Count())].Value;
        }
        url = $"{uri.Scheme}://{agentService.Address}:{agentService.Port}{uri.PathAndQuery}";
        string content = await _httpSender.InvokeApi(url);
        #endregion
        return View(JsonConvert.DeserializeObject<CustomerUser>(content));
    }
}

public class HttpSender
{
    public async Task<string> InvokeApi(string url)
    {
        using (HttpClient client = new HttpClient())
        {
            HttpRequestMessage message = new HttpRequestMessage();
            message.Method = HttpMethod.Get;
            message.RequestUri = new Uri(url);
            var result = client.SendAsync(message).Result;
            string content = result.Content.ReadAsStringAsync().Result;
            return content;
        }
    }
}

启动web项目,访问User-Index这个视图,会轮询不同的服务实例。

但是这样做不好，客户端都需要和Consul进行连接,拿到所有的服务实例,直接和服务实例进行交互,服务实例就暴露了--所以需要网关。

网关将服务实例与客户端进行隔离,是所有Api请求的入口。因此可以统一鉴权。当然微服务网关的作用有很多,大家可自行百度了解。

新建一个网关的项目,请求先到达网关,再由网关分发请求到不同的实例。如图：

 Consul整合GeteWay.引入NuGet包：Ocelot、Ocelot.Provider.Consul

修改Startup类：

public class Startup
{
    public Startup(IConfiguration configuration)
    {
        Configuration = configuration;
    }

    public IConfiguration Configuration { get; }

    // This method gets called by the runtime. Use this method to add services to the container.
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddOcelot().AddConsul();
        //services.AddControllers();
    }

    // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
    public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
    {
        //将默认的请求管道全部丢掉
        app.UseOcelot();
        //if (env.IsDevelopment())
        //{
        //    app.UseDeveloperExceptionPage();
        //}

        //app.UseHttpsRedirection();

        //app.UseRouting();

        //app.UseAuthorization();

        //app.UseEndpoints(endpoints =>
        //{
        //    endpoints.MapControllers();
        //});
    }
}

修改Program类：

public class Program
{
    public static void Main(string[] args)
    {
        CreateHostBuilder(args).Build().Run();
    }

    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
        .ConfigureAppConfiguration(conf =>
        {
            conf.AddJsonFile("configuration.json", optional: false,
                reloadOnChange: true);
        })
            .ConfigureWebHostDefaults(webBuilder =>
            {
                webBuilder.UseStartup<Startup>();
            });
}

新建一个configuration.json的文件,用于配置策略等...

Routes中路由规则可以配置多个

{//*************************单地址多实例负载均衡+Consul*****************************
  "Routes": [
    {
      //GeteWay转发=>Downstream
      "DownstreamPathTemplate": "/api/{url}", //服务地址--url变量
      "DownstreamScheme": "http",
      //http://localhost:6299/T5/User/GetCustomerUser
      "UpstreamPathTemplate": "/T5/{url}", //网关地址--url变量 冲突的还可以加权重Priority
      "UpstreamHttpMethod": [ "Get", "Post" ],
      "UseServiceDiscovery": true, //使用服务发现
      "ServiceName": "MicroserviceAttempt", //Consul服务名称
      "LoadBalancerOptions": {
        "Type": "RoundRobin" //轮询  //"LeastConnection":最少连接数服务器   "NoloadBalance":不负载均衡     "CookieStickySession":会话粘滞
      }
    }
  ],
  "GlobalConfiguration": {
    "BaseUrl": "http://127.0.0.1:6299",
    "ServiceDiscoveryProvider": {
      "Host": "localhost",
      "Port": 8500,
      "Type": "Consul"//由Consul提供服务发现,每次请求去Consul
    }
    //"ServiceDiscoveryProvider": {
    //  "Host": "localhost",
    //  "Port": 8500,
    //  "Type": "PollConsul", //由Consul提供服务发现,每次请求去Consul
    //  "PollingInterval": 1000//轮询Consul,评率毫秒--down是不知道的
    //}
  }
  //*************************单地址多实例负载均衡+Consul*****************************
}

启动网关(项目)服务：

dotnet GateWay-Ocelot.dll --urls="http://*:6299" --ip="127.0.0.1" --port=6299

调用服务接口：

每请求一次就轮询不同的服务实例,达到负载均衡。

服务治理、熔断、降级

服务治理-缓存

引入NuGet包：Ocelot.Cache.Cache

修改ConfigureServices方法：

public void ConfigureServices(IServiceCollection services)
{
    services.AddOcelot().AddConsul()
        .AddCacheManager(m =>
        {
            m.WithDictionaryHandle();//默认字典存储
        });
    //services.AddControllers();
}

修改configuration.json文件

"Routes": [
    {
      //GeteWay转发=>Downstream
      "DownstreamPathTemplate": "/api/{url}", //服务地址--url变量
      "DownstreamScheme": "http",
      //http://localhost:6299/T5/User/GetCustomerUser
      "UpstreamPathTemplate": "/T5/{url}", //网关地址--url变量 冲突的还可以加权重Priority
      "UpstreamHttpMethod": [ "Get", "Post" ],
      "UseServiceDiscovery": true, //使用服务发现
      "ServiceName": "MicroserviceAttempt", //Consul服务名称
      "LoadBalancerOptions": {
        "Type": "RoundRobin" //轮询  //"LeastConnection":最少连接数服务器   "NoloadBalance":不负载均衡     "CookieStickySession":会话粘滞
      },
      //使用缓存
      "FileCacheOptions": {
        "TtlSeconds": 15,//过期时间
        "Region": "UserCache" //可以调用Api清理
      }
    }
  ]

再次调用会发现每隔15秒数据才会变.

自定义缓存

新建类CustomeCache

/// <summary>
/// 自定义Cache
/// </summary>
public class CustomeCache : IOcelotCache<CachedResponse>
{
    public class CacheDataModel
    {
        public CachedResponse CachedResponse { get; set; }
        public DateTime TimeOut { get; set; }
        public string Region { get; set; }
    }

    private static Dictionary<string, CacheDataModel> keyValuePairs = 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)
    {
        Console.WriteLine($"调用了{nameof(CustomeCache)}--{nameof(Add)}");
        keyValuePairs[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)
    {
        Console.WriteLine($"调用了{nameof(CustomeCache)}--{nameof(AddAndDelete)}");
        keyValuePairs[key] = new CacheDataModel()
        {
            CachedResponse = value,
            Region = region,
            TimeOut = DateTime.Now.Add(ttl)
        };
    }

    /// <summary>
    /// 清除
    /// </summary>
    /// <param name="region"></param>
    public void ClearRegion(string region)
    {
        Console.WriteLine($"调用了{nameof(CustomeCache)}--{nameof(ClearRegion)}");
        var keyList = keyValuePairs.Where(m => m.Value.Region.Equals(region)).Select(e => e.Key);
        foreach (var item in keyList)
        {
            keyValuePairs.Remove(item);
        }
    }

    /// <summary>
    /// 获取
    /// </summary>
    /// <param name="key"></param>
    /// <param name="region"></param>
    /// <returns></returns>
    public CachedResponse Get(string key, string region)
    {
        Console.WriteLine($"调用了{nameof(CustomeCache)}--{nameof(Get)}");
        if (keyValuePairs.ContainsKey(key) && keyValuePairs[key] != null && keyValuePairs[key].TimeOut > DateTime.Now && keyValuePairs[key].Region.Equals(region))
            return keyValuePairs[key].CachedResponse;
        else
            return null;
    }
}

在ConfigureServices方法中加入：

public void ConfigureServices(IServiceCollection services)
{
    services.AddOcelot().AddConsul()
        .AddCacheManager(m =>
        {
            m.WithDictionaryHandle();//默认字典存储
        });
    //services.AddControllers();
    //这里的IOcelotCache<CachedResponse>是默认缓存的约束--准备替换成自定义的
    services.AddSingleton<IOcelotCache<CachedResponse>, CustomeCache>();
}

调用，15秒刷新一次。

 雪崩效应：微服务架构下,单个服务的故障而引发系列服务故障。

解决：1.超时：调用服务的操作可以配置为执行超时,如果服务未能在这个给定时间内响应,将回复一个失败的消息。

           2.熔断：使用断路器来检测故障是否已得到解决,防止请求反复尝试执行一个可能会失败的操作,从而减少等待纠正故障的时间。

安装NuGet包：Ocelot.Provider.Polly

修改ConfigureServices方法：

services.AddOcelot().AddConsul()
    .AddCacheManager(m =>
    {
        m.WithDictionaryHandle();//默认字典存储
    })
    .AddPolly();
//services.AddControllers();

//这里的IOcelotCache<CachedResponse>是默认缓存的约束--准备替换成自定义的
services.AddSingleton<IOcelotCache<CachedResponse>, CustomeCache>();

修改configuration.json文件

  "Routes": [
    {
      //GeteWay转发=>Downstream
      "DownstreamPathTemplate": "/api/{url}", //服务地址--url变量
      "DownstreamScheme": "http",
      //http://localhost:6299/T5/User/GetCustomerUser
      "UpstreamPathTemplate": "/T5/{url}", //网关地址--url变量 冲突的还可以加权重Priority
      "UpstreamHttpMethod": [ "Get", "Post" ],
      "UseServiceDiscovery": true, //使用服务发现
      "ServiceName": "MicroserviceAttempt", //Consul服务名称
      "LoadBalancerOptions": {
        "Type": "RoundRobin" //轮询  //"LeastConnection":最少连接数服务器   "NoloadBalance":不负载均衡     "CookieStickySession":会话粘滞
      },
      //使用缓存
      "FileCacheOptions": {
        "TtlSeconds": 15, //过期时间
        "Region": "UserCache" //可以调用Api清理
      },
      "QoSOptions": {
        "ExceptionsAllowedBeforeBreaking": 3, //熔断之前允许多少个异常请求
        "DurationOfBreak": 10000, //熔断的时间,单位为ms.超过这个时间可再请求
        "TimeoutValue": 4000 //如果下游请求的处理时间超过多少则将请求设置为超时  默认90秒
      }
    }
  ],

故意设置接口休眠5秒钟

调用：

 限流：限制单位时间内的请求数(保护机制),超过就返回指定信息。

修改configuration.json文件

  "Routes": [
    {
      //GeteWay转发=>Downstream
      "DownstreamPathTemplate": "/api/{url}", //服务地址--url变量
      "DownstreamScheme": "http",
      //http://localhost:6299/T5/User/GetCustomerUser
      "UpstreamPathTemplate": "/T5/{url}", //网关地址--url变量 冲突的还可以加权重Priority
      "UpstreamHttpMethod": [ "Get", "Post" ],
      "UseServiceDiscovery": true, //使用服务发现
      "ServiceName": "MicroserviceAttempt", //Consul服务名称
      "LoadBalancerOptions": {
        "Type": "RoundRobin" //轮询  //"LeastConnection":最少连接数服务器   "NoloadBalance":不负载均衡     "CookieStickySession":会话粘滞
      },
      //使用缓存
      "FileCacheOptions": {
        "TtlSeconds": 15, //过期时间
        "Region": "UserCache" //可以调用Api清理
      },
      //限流  张队长贡献的
      "RateLimitOptions": {
        "ClientWhitelist": ["Microservice","Attempt"],//白名单  ClientId区分大小写
        "EnableRateLimiting": true,
        "Period": "1s", //5m 1h 1d
        "PeriodTimespan": 30,//多少秒之后客户端可以重试
        "Limit": 5 //统计时间段内允许的最大请求数
      },
      "QoSOptions": {
        "ExceptionsAllowedBeforeBreaking": 3, //熔断之前允许多少个异常请求
        "DurationOfBreak": 10000, //熔断的时间,单位为ms.超过这个时间可再请求
        "TimeoutValue": 4000 //如果下游请求的处理时间超过多少则将请求设置为超时  默认90秒
      }
    }
  ],
  "GlobalConfiguration": {
    "BaseUrl": "http://127.0.0.1:6299",
    "ServiceDiscoveryProvider": {
      "Host": "localhost",
      "Port": 8500,
      "Type": "Consul" //由Consul提供服务发现,每次请求去Consul
    },
    "RateLimitOptions": {
      "QuotaExceededMessage": "Customize Tips!", //限流时返回的消息
      "HttpStatusCode": 999 //限流时返回的code
    }
    //"ServiceDiscoveryProvider": {
    //  "Host": "localhost",
    //  "Port": 8500,
    //  "Type": "PollConsul", //由Consul提供服务发现,每次请求去Consul
    //  "PollingInterval": 1000//轮询Consul,评率毫秒--down是不知道的
    //}
  }

调用接口,超过五次就会限流：

 当设置了白名单后,就对来访的请求就不做限流机制

到此就结尾了。如有不当的地方,请谅解,希望能帮到大家。

代码已上传至我的github:

https://github.com/Davenever/MicroserviceBegin.git