ASP.NET CORE 使用Consul实现服务治理与健康检查(2)——源码篇

题外话

笔者有个习惯，就是在接触新的东西时，一定要先搞清楚新事物的基本概念和背景，对之有个相对全面的了解之后再开始进入实际的编码，这样做最主要的原因是尽量避免由于对新事物的认知误区导致更大的缺陷，bug 一旦发生，将比普通的代码缺陷带来更加昂贵的修复成本。

相信有了前一篇和园子里其他同学的文章，你已经基本上掌握了使用 consul 所需要具备的背景知识，那么就让我们来看下，具体到 asp.net core 中，如何更加优雅的编码。

consul 在 asp.net core 中的使用

consul服务在注册时需要注意几个问题：

1. 那就是必须是在服务完全启动之后再进行注册，否则可能导致服务在启动过程中已经注册到 consul server，这时候我们要利用 iapplicationlifetime 应用程序生命周期管理中的 applicationstarted 事件。
2. 应用程序向 consul 注册时，应该在本地记录应用 id，以此解决每次重启之后，都会向 consul 注册一个新实例的问题，便于管理。

具体代码如下：

注意：以下均为根据排版要求所展示的示意代码，并非完整的代码

1. 服务治理之服务注册

• 1.1 服务注册扩展方法

public static iapplicationbuilder agentserviceregister(this iapplicationbuilder app,
    iapplicationlifetime lifetime,
    iconfiguration configuration,
    iconsulclient consulclient,
    ilogger logger)
{
    try
    {
        var urlsconfig = configuration["server.urls"];
        argumentcheck.notnullorwhitespace(urlsconfig, "未找到配置文件中关于 server.urls 相关配置！");

        var urls = urlsconfig.split(';');
        var port =  urls.first().substring(httpurl.lastindexof(":") + 1);
        var ip = getprimaryipaddress(logger);
        var registrationid = getregistrationid(logger);

        var servicename = configuration["apollo:appid"];
        argumentcheck.notnullorwhitespace(servicename, "未找到配置文件中 apollo:appid 对应的配置项！");

        //程序启动之后注册
        lifetime.applicationstarted.register(() =>
        {
            var healthcheck = new agentservicecheck
            {
                deregistercriticalserviceafter = timespan.fromseconds(5),
                interval = 5,
                http = $"http://{ip}:{port}/health",
                timeout = timespan.fromseconds(5),
                tlsskipverify = true
            };

            var registration = new agentserviceregistration
            {
                checks = new[] { healthcheck },
                id = registrationid,
                name = servicename.tolower(),
                address = ip,
                port = int.parse(port),
                tags = ""//手动高亮
            };

            consulclient.agent.serviceregister(registration).wait();
            logger.loginformation($"服务注册成功! 注册地址：{((consulclient)consulclient).config.address}, 注册信息：{registration.tojson()}");
        });

        //优雅的退出
        lifetime.applicationstopping.register(() =>
        {
            consulclient.agent.servicederegister(registrationid).wait();
        });

        return app;
    }
    catch (exception ex)
    {
        logger?.logspider(loglevel.error, "服务发现注册失败！", ex);
        throw ex;
    }
}

private static string getprimaryipaddress(ilogger logger)
{
    string output = getlocalipaddress();
    logger?.loginformation(loglevel.information, "获取本地网卡地址结果：{0}", output);

    if (output.length > 0)
    {
        var ips = output.split(new[] { ',' }, stringsplitoptions.removeemptyentries);
        if (ips.length == 1) return ips[0];
        else
        {
            var localips = ips.where(w => w.startswith("10"));//内网网段
            if (localips.count() > 0) return localips.first();
            else return ips[0];
        }
    }
    else
    {
        logger?.logspider(loglevel.error, "没有获取到有效的ip地址，无法注册服务到服务中心！");
        throw new exception("获取本机ip地址出错，无法注册服务到注册中心！");
    }
}

public static string getlocalipaddress()
{
    if (!string.isnullorwhitespace(_localipaddress)) return _localipaddress;

    string output = "";
    try
    {
        foreach (networkinterface item in networkinterface.getallnetworkinterfaces())
        {
            if (item.operationalstatus != operationalstatus.up) continue;

            var adapterproperties = item.getipproperties();
            if (adapterproperties.gatewayaddresses.count == 0) continue;

            foreach (unicastipaddressinformation address in adapterproperties.unicastaddresses)
            {
                if (address.address.addressfamily != addressfamily.internetwork) continue;
                if (ipaddress.isloopback(address.address)) continue;

                output = output += address.address.tostring() + ",";
            }
        }
    }
    catch (exception e)
    {
        console.writeline("获取本机ip地址失败！");
        throw e;
    }

    if (output.length > 0)
        _localipaddress = output.trimend(',');
    else
        _localipaddress = "unknown";

    return _localipaddress;
}

private static string getregistrationid(ilogger logger)
{
    try
    {
        var basepath = directory.getcurrentdirectory();
        var folderpath = path.combine(basepath, "registrationid");
        if (!directory.exists(folderpath))
            directory.createdirectory(folderpath);

        var path = path.combine(basepath, "registrationid", ".id");
        if (file.exists(path))
        {
            var lines = file.readalllines(path, encoding.utf8);
            if (lines.count() > 0 && !string.isnullorempty(lines[0]))
                return lines[0];
        }

        var id = guid.newguid().tostring();
        file.appendalllines(path, new[] { id });
        return id;
    }
    catch (exception e)
    {
        logger?.logwarning(e, "获取 registration id 错误");
        return guid.newguid().tostring();
    }
}

• 1.2 健康检查中间件

既然健康检查是通过http请求来实现的，那么我们可以通过 healthmiddleware 中间件来实现：

public static void usehealth(this iapplicationbuilder app)
{
    app.usemiddleware<healthmiddleware>();
}

public class healthmiddleware
{
    private readonly requestdelegate _next;
    private readonly string _healthpath = "/health";

    public healthmiddleware(requestdelegate next, iconfiguration configuration)
    {
        this._next = next;
        var healthpath = configuration["consul:healthpath"];
        if (!string.isnullorempty(healthpath))
        {
            this._healthpath = healthpath;
        }
    }

    //监控检查可以返回更多的信息，例如服务器资源信息
    public async task invoke(httpcontext httpcontext)
    {
        if (httpcontext.request.path == this._healthpath)
        {
            httpcontext.response.statuscode = (int)httpstatuscode.ok;
            await httpcontext.response.writeasync("i'm ok!");
        }
        else
            await this._next(httpcontext);
    }
}

• 1.3 startup 配置

public void configureservices(iservicecollection services)
{
    services.addmvc().setcompatibilityversion(compatibilityversion.version_2_2);
    //手动高亮
    services.addsingleton<iconsulclient>(sp =>
    {
        argumentcheck.notnullorwhitespace(this.configuration["consul:address"], "未找到配置中consul:address对应的配置");
        return new consulclient(c => { c.address = new uri(this.configuration["consul:address"]); });
    });
}
public void configure(iapplicationbuilder app, ihostingenvironment env, iapplicationlifetime lifetime, iconsulclient consulclient, ilogger<startup> logger)
{
    ...
    app.usehealth();//手动高亮
    app.usemvc();
    app.agentserviceregister(lifetime, this.configuration, consulclient, logger);//手动高亮
}

2. 服务治理之服务发现

public class servicemanager : iservicemanager
{
    private readonly ihttpclientfactory _httpclientfactory;
    private readonly ilogger _logger;
    private readonly iconsulclient _consulclient;
    private readonly ilist<func<strategydelegate, strategydelegate>> _components;
    private strategydelegate _strategy;

    public servicemanager(ihttpclientfactory httpclientfactory,
        iconsulclient consulclient,
        ilogger<servicemanager> logger)
    {
        this._cancellationtokensource = new cancellationtokensource();
        this._components = new list<func<strategydelegate, strategydelegate>>();

        this._httpclientfactory = httpclientfactory;
        this._optionsconsulconfig = optionsconsulconfig;
        this._logger = logger;
        this._consulclient = consulclient;
    }

    public async task<httpclient> gethttpclientasync(string servicename, string erroripaddress = null, string hashkey = null)
    {
        //重要：获取所有健康的服务
        var resonse = (await this._consulclient.health.service(servicename.tolower(), this._cancellationtokensource.token)).response;
        var filteredservice = this.getservicenode(servicename, resonse.toarray(), hashkey);
        return this.createhttpclient(servicename.tolower(), filteredservice.service.address, filteredservice.service.port);
    }

    private serviceentry getservicenode(string servicename, serviceentry[] services, string hashkey = null)
    {
        if (this._strategy == null)
        {
            lock (this) { if (this._strategy == null) this._strategy = this.build(); }
        }

        //策略过滤
        var filterservice = this._strategy(servicename, services, hashkey);
        return filterservice.firstordefault();
    }

    private httpclient createhttpclient(string servicename, string address, int port)
    {
        var httpclient = this._httpclientfactory.createclient(servicename);
        httpclient.baseaddress = new system.uri($"http://{address}:{port}");
        return httpclient;
    }
}

服务治理之——访问策略

服务在注册时，可以通过配置或其他手段给当前服务配置相应的 tags ，同样在服务获取时，我们也将同时获取到该服务的 tags, 这对于我们实现策略访问夯实了基础。例如开发和测试共用一套服务注册发现基础设施（当然这实际不可能），我们就可以通过给每个服务设置环境 tag ，以此来实现环境隔离的访问。这个 tag 维度是没有限制的，开发人员完全可以根据自己的实际需求进行打标签，这样既可以通过内置默认策略兜底，也允许开发人员在此基础之上动态的定制访问策略。

笔者所实现的访问策略方式类似于 asp.net core middleware 的方式，并且笔者认为这个设计非常值得借鉴，并参考了部分源码实现,使用方式也基本相同。

源码实现如下：

//策略委托
public delegate serviceentry[] strategydelegate(string servicename, serviceentry[] services, string hashkey = null);

//服务管理
public class servicemanager:iservicemanager
{
    private readonly ilist<func<strategydelegate, strategydelegate>> _components;
    private strategydelegate _strategy;//策略链

    public servicemanager()
    {
        this._components = new list<func<strategydelegate, strategydelegate>>();
    }

    //增加自定义策略
    public iservicemanager usestrategy(func<strategydelegate, strategydelegate> strategy)
    {
        _components.add(strategy);
        return this;
    }

    //build 最终策略链
    private strategydelegate build()
    {
        strategydelegate strategy = (sn, services, key) =>
        {
            return new defaultstrategy().invoke(null, sn, services, key);
        };

        foreach (var component in _components.reverse())
        {
            strategy = component(strategy);
        }

        return strategy;
    }
}
public class defaultstrategy : istrategy
{
    private ushort _idx;
    public defaultstrategy(){}

    public serviceentry[] invoke(strategydelegate next, string servicename, serviceentry[] services, string hashkey = null)
    {
        var service = services.length == 1 ? services[0] : services[this._idx++ % services.length];
        var result = new[] { service };
        return next != null ? next(servicename, result, hashkey) : result;
    }
}

自定义策略扩展方法以及使用

public static iapplicationbuilder usestrategy(this iapplicationbuilder app)
{
    var servicemanager = app.applicationservices.getrequiredservice<iservicemanager>();
    var strategies = app.applicationservices.getservices<istrategy>();

    //注册所有的策略
    foreach (var strategy in strategies)
    {
        servicemanager.usestrategy(next =>
        {
            return (servicename, services, hashkey) => strategy.invoke(next, servicename, services, hashkey);
        });
    }
    return app;
}

public class startup
{
    public void configureservices(iservicecollection services)
    {
        ...
        services.addsingleton<istrategy, customstrategy>(); //自定义策略1
        services.addsingleton<istrategy, customstrategy2>(); //自定义测率2
    }

    public void configure(iapplicationbuilder app, ihostingenvironment env, iapplicationlifetime lifetime, iconsulclient consulclient, ilogger<startup> logger)
    {
        app.usestrategy(); //手动高亮
        app.agentserviceregister(lifetime, this.configuration, consulclient, logger);
    }
}