.NetCore下ES查询驱动 PlainElastic .Net 升级官方驱动 Elasticsearch .Net

程序员文章站 2022-03-11 08:40:51

1.背景由于历史原因，笔者所在的公司原有的ES查询驱动采用的是 PlainElastic.Net, 经过询问原来是之前PlainElastic.Net在园子里文档较多，上手比较容易,所以最初作者选用了该驱动，而发布也由于历史原因都部署在 windows 服务器上，基于 .NET Framework ......

1.背景

由于历史原因，笔者所在的公司原有的es查询驱动采用的是 plainelastic.net, 经过询问原来是之前plainelastic.net在园子里文档较多，上手比较容易,所以最初作者选用了该驱动，而发布也由于历史原因都部署在 windows 服务器上，基于 .net framework开发。

后来由于迁移 .net core 平台的需要，对代码进行了升级，同时部署平台也迁移至 centos7 服务器，升级过程比较顺利，由于没有使用特殊api，所以几乎没有对业务代码做更多的修改，同时测试阶段由于没有多余的机器，仍然放在了原有windows服务器上做的测试，一切都没有问题，完美上线。

事发突然，某天接到运维部门反馈，部署查询服务的机器突然出现 tcp 连接数超高的问题，同时这台机器其他的tcp服务也无法建立新的连接，但已经建立的连接不受影响。联想到 elasticsearch 查询服务是基于http 请求的，脑子里马上联想到 .net core 下 httpclient 如果每次访问都创建新实例，则会每次都建立新的tcp连接，而 linux 对已释放端口回收的时间窗口，会导致在高并发情况下，客户端机器端口占用持续增加，同时被调用服务端连接数也会持续增加。

基于此猜测，立马去扒了一下plainelastic.net源代码：

.NetCore下ES查询驱动 PlainElastic .Net 升级官方驱动 Elasticsearch .Net

源码地址：https://github.com/yegoroff/plainelastic.net/blob/master/src/plainelastic.net/connection/elasticconnection.cs
果然如猜测的那样，每次都创建了新的 httpwebrequest 实例，看了作者的最后维护时间也已经是3年前了，可能是后来官方驱动日趋完善，作者也便停止了维护。

既然如此，那么让我们看下官方最新驱动源码是否如我们想象，是基于httpclientfactory来解决这个问题的？
.NetCore下ES查询驱动 PlainElastic .Net 升级官方驱动 Elasticsearch .Net

源码地址：https://github.com/elastic/elasticsearch-net/blob/master/src/elasticsearch.net/connection/httpconnection.cs

上述代码看来，官方驱动并非是采用微软官方建议的 httpclientfactory ，而是官方底层自己维护的一个线程安全的字典来管理 httpclient 实例池，虽是自己实现，但效果一样：相同地址的请求，是链接复用的，这样就解决不断开启 tcp 连接的问题。

问题找到，立马进行驱动升级:

2.驱动升级

说明： elasticsearch.net官方驱动地址：

官方驱动分为 low level client 和 nest(heigh level client),其中low level client 仅仅做了最基本的封装，几乎等价于http原生调用，带来了极大的灵活性的同时，也带来使用成本，而对于开发人员来说使用 nest 提供的更加高级的api，可以更加快速的进行开发工作，也同时可以利用到 .net 所提供的各种语法糖，比如 => 表达式。

话不多说，看示例代码：

实例创建

public elasticservice()
{
    var uris = new uri[] { new uri("http://172.17.78.111:9200"), new uri("http://172.17.78.112:9200") }; //支持多个节点
    var connectionpool = new sniffingconnectionpool(uris);
    var settings = new connectionsettings(connectionpool).defaultindex("testindex");//注意index不可以大写
    settings.basicauthentication("", ""); //设置账号密码，没有可以跳过
    this._client = new elasticclient(settings);
}

插入待测试数据

public class people 
{
    public guid id { get; set; }
    public string name { get; set; }
    public int age { get; set; }
    public datetime birthday { get; set; }
    public bool gender { get; set; }
    public string address { get; set; }
    public datetime createtime { get; set; } = datetime.now;
}

//批量插入
public async task<ibulkresponse> addpeopleasync(people[] peoples)
{
    var descriptor = new bulkdescriptor();
    foreach (var p in peoples)
    {
        var response = await _client.indexdocumentasync(p);
        descriptor.index<people>(op => op.document(p));
    }
    return await _client.bulkasync(descriptor);//批量插入
}

多查询条件拼接

public querycontainer buildquerycontainer(searchcondition condition)
{
    var querycombin = new list<func<querycontainerdescriptor<people>, querycontainer>>();
    if (!string.isnullorempty(condition.name))
        querycombin.add(mt => mt.match(m => m.field(t => t.name).query(condition.name))); //字符串匹配

    if (condition.age.hasvalue)
        querycombin.add(mt => mt.range(m => m.field(t => t.address).greaterthanorequals(condition.age))); //数值区间匹配

    if (!string.isnullorempty(condition.address))
        querycombin.add(mt => mt.matchphrase(m => m.field(t => t.address).query(condition.address))); //短语匹配

    if (!condition.gender.hasvalue)
        querycombin.add(mt => mt.term(m => m.field(t => t.gender).value(condition.gender)));//精确匹配

    return query<people>.bool(b => b
        .must(querycombin)
        .filter(f => f
            .daterange(dr => dr.field(t => t.createtime) //时间范围匹配
                .greaterthanorequals(datemath.anchored(condition.begincreatetime.tostring("yyyy-mm-ddthh:mm:ss")))
                .lessthanorequals(datemath.anchored(condition.endcreatetime.tostring("yyyy-mm-ddthh:mm:ss"))))));
}

提示：match 和 matchphrase 的区别，例如对于"长宁区"

match 会将"长宁区"进行分词匹配，例如只要包含"区"的数据（比如静安区），也会被查询命中

matchphrase 则可以理解为短语匹配，只有当数据包含“长宁区”完整短语的数据，才会被查询命中

增加分页查询接口

public async task<pagedresult<people[]>> querypeopleasync(searchcondition condition, int pageindex, int pagesize)
{
    var query = this.buildquerycontainer(condition);
    var response = await this._client.searchasync<people>(s => s
            .index("testindex")
            .from(pageindex * pagesize)
            .size(pagesize)
            .query(q => query)
            .sort(st => st.descending(d => d.createtime)));

    if (response.apicall.success)
    {
        return new pagedresult<people[]>
        {
            pageindex = pageindex,
            pagesize = pagesize,
            total = response.total,
            returnobj = response.hits.select(s => s.source).toarray()
        };
    }

    return new pagedresult<people[]> { issuccess = false };
}

编写单元测试

[testmethod]
public async task querypeopletest()
{
    var condition = new searchcondition
    {
        address="长宁区",
        begincreatetime = datetime.now.adddays(-1),
        endcreatetime = datetime.now
    };

    var result = await this._elasticservice.querypeopleasync(condition, 0, 3);
    assert.istrue(result.issuccess);
}