ES Client性能测试初探

FunTester 2023-03-28 原文

最近在工作中协助研发进行了ES优化，效果还是非常明显的，几乎翻倍。除了通过各种业务接口测试ES性能以外，还可以直接请求ES接口，绕过服务，这样应该数据回更加准确。所以，ES Client学起来。

准备工作

首先，先准备了一个ES服务，这里就不多赘述了，大家自己在尝试的时候一定主意好ES Server和ES Client的版本要一致。其次，新建项目，添加依赖。

学习资料

搜一下，能搜到很多的ES学习资料，建议先去看看大厂出品的基础知识了解一下ES功能。然后就可以直接看ES的API了。下面是ES官方的文档地址： https://www.elastic.co/guide/en/elasticsearch/client/java-rest/6.7/java-rest-high-search.html

如果能能查看自己公司项目源码的小伙伴可以多研究研发的代码，能够更好结合业务理解ES API的使用。

ES Client

HTTP请求

这里说一下，很多ES查询功能都是通过HTTP请求完成的，GET请求，body传参，一开始还是比较懵逼的。查了一些资料需要自己实现是个body携带数据的HTTPGET请求，下面是我的实现代码：

package com.funtester.httpclient

import org.apache.http.client.methods.HttpEntityEnclosingRequestBase

import javax.annotation.concurrent.NotThreadSafe

/**
 * HttpGet请求携带body参数
 */
@NotThreadSafe
class HttpGetByBody extends HttpEntityEnclosingRequestBase {

    static final String METHOD_NAME = "GET";

    /**
     * 获取方法（必须重载）
     *
     * @return
     */
    @Override
    String getMethod() {
        return METHOD_NAME;
    }

    /**
     * PS:不能照抄{@link org.apache.http.client.methods.HttpPost}
     * @param uri
     */
    HttpGetByBody(final String uri) {
        this(new URI(uri))
    }

    HttpGetByBody(final URI uri) {
        super();
        setURI(uri);
    }

    HttpGetByBody() {
        super();
    }
}

ES Client

如果使用HTTP接口进行ES操作，需要组合多层级的参数，这个写起来会比较麻烦、可读性也比较差，而且更加容易出错。所以，还是使用ES Client作为操作ES的基础框架。

如果翻看ES Client源码，最终也是通过HttpClient发起HTTP请求的，这中间进行了很多的封装。这里分享一下ES Client的HTTP Client创建代码部分：

    private CloseableHttpAsyncClient createHttpClient() {
        //default timeouts are all infinite
        RequestConfig.Builder requestConfigBuilder = RequestConfig.custom()
                .setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_MILLIS)
                .setSocketTimeout(DEFAULT_SOCKET_TIMEOUT_MILLIS);
        if (requestConfigCallback != null) {
            requestConfigBuilder = requestConfigCallback.customizeRequestConfig(requestConfigBuilder);
        }

        try {
            HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClientBuilder.create().setDefaultRequestConfig(requestConfigBuilder.build())
                //default settings for connection pooling may be too constraining
                .setMaxConnPerRoute(DEFAULT_MAX_CONN_PER_ROUTE).setMaxConnTotal(DEFAULT_MAX_CONN_TOTAL)
                .setSSLContext(SSLContext.getDefault())
                .setTargetAuthenticationStrategy(new PersistentCredentialsAuthenticationStrategy());
            if (httpClientConfigCallback != null) {
                httpClientBuilder = httpClientConfigCallback.customizeHttpClient(httpClientBuilder);
            }

            final HttpAsyncClientBuilder finalBuilder = httpClientBuilder;
            return AccessController.doPrivileged(new PrivilegedAction<CloseableHttpAsyncClient>() {
                @Override
                public CloseableHttpAsyncClient run() {
                    return finalBuilder.build();
                }
            });
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("could not create the default ssl context", e);
        }
    }

可以看出ES Client用到了HttpClient的异步Client，我猜是用future实现同步返回响应结果，这个没仔细看，有错请指出。这里也回答我的自己的一个疑惑，ES Client是支持并发的。

ES Client 封装

就我自己的观察，ES Client的封装程度非常高，完全可以拿来就用。我担心自己过几天之后就不知道改怎么用这些ES Client 的API了，所以又进行了一次封装，权当是一个学习笔记类。

封装代码有点多，放到了文末。

测试用例

添加数据

这个可以用来跑一部分数据到ES里。

package com.funtest.groovytest

import com.alibaba.fastjson.JSONObject
import com.funtester.es.ESClient
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunQpsConcurrent

import java.util.concurrent.atomic.AtomicInteger

class ESC extends SourceCode {

    static void main(String[] args) {
        def client = new ESClient("127.0.0.1", 9200, "http")
        def data = new JSONObject()
        data.name = "FunTester"
        data.age = getRandomInt(100)
        def index = new AtomicInteger(0)
        def test = {
            data.put("time", index.getAndIncrement())
            client.index("fun", "tt", data)
        }
        new FunQpsConcurrent(test, "ES添加数据").start()
    }

}

如果想测试添加、删除功能，只需要把test闭包内容修改即可。

        def test = {
            data.put("time", index.getAndIncrement())
            client.delete("fun", "tt", client.index("fun", "tt", data))
        }

下面是搜索功能的性能测试用例：

package com.funtest.groovytest

import com.alibaba.fastjson.JSONObject
import com.funtester.es.ESClient
import com.funtester.frame.SourceCode
import com.funtester.frame.execute.FunQpsConcurrent
import org.elasticsearch.index.query.QueryBuilders

import java.util.concurrent.atomic.AtomicInteger

class ESC extends SourceCode {

    static void main(String[] args) {
        def client = new ESClient("127.0.0.1", 9200, "http")
        def data = new JSONObject()
        data.name = "FunTester"
        data.age = getRandomInt(100)
        def index = new AtomicInteger(0)
        def test = {
            client.search("fun", QueryBuilders.matchQuery("time", getRandomInt(10)))
        }
        new FunQpsConcurrent(test, "ES搜索").start()
    }

}

ES Client API封装类

package com.funtester.es

import com.funtester.frame.SourceCode
import groovy.util.logging.Log4j2
import org.apache.http.HttpHost
import org.elasticsearch.action.delete.DeleteRequest
import org.elasticsearch.action.get.GetRequest
import org.elasticsearch.action.get.GetResponse
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.index.IndexResponse
import org.elasticsearch.action.search.SearchRequest
import org.elasticsearch.action.search.SearchResponse
import org.elasticsearch.action.search.SearchScrollRequest
import org.elasticsearch.client.RequestOptions
import org.elasticsearch.client.RestClient
import org.elasticsearch.client.RestHighLevelClient
import org.elasticsearch.common.unit.TimeValue
import org.elasticsearch.index.query.QueryBuilder
import org.elasticsearch.search.SearchHits
import org.elasticsearch.search.builder.SearchSourceBuilder
import org.elasticsearch.search.fetch.subphase.FetchSourceContext

import java.util.concurrent.TimeUnit

/**
 * ES客户端API练习类
 */
@Log4j2
class ESClient extends SourceCode {

    String host

    int port

    String scheme

    RestHighLevelClient client

    ESClient(String host, int port = 9200, String scheme = "http") {
        this.host = host
        this.port = port
        this.scheme = scheme
        // 设置验证信息，填写账号及密码
        //        CredentialsProvider credentialsProvider = new BasicCredentialsProvider()
        //        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("user", "passwd"))
        def builder = RestClient.builder(new HttpHost(host, port, scheme))
        // 设置认证信息
        //        builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
        //
        //            @Override
        //            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
        //                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
        //            }
        //        })
        builder.setMaxRetryTimeoutMillis(1000)
        client = new RestHighLevelClient(builder)
    }

    /**
     * 添加数据
     * @param index
     * @param type
     * @param data
     * @return
     */
    def index(String index, type, Map data) {
        IndexRequest indexRequest = new IndexRequest(index, type).source(data)
        IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT)
        indexResponse.getId()
    }

    /**
     * 获取数据
     * @param index
     * @param type
     * @param id
     * @return
     */
    def get(String index, type, id) {
        // 查询文档
        GetRequest getRequest = new GetRequest(index, type, id)
        GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT)
        if (getResponse.isExists()) {
            getResponse.getSourceAsString()
        }
    }

    /**
     * 数据是否存在
     * @param index
     * @param type
     * @param id
     * @return
     */
    def exists(String index, type, id) {
        GetRequest getRequest = new GetRequest(index, type, id)
        getRequest.fetchSourceContext(new FetchSourceContext(false))
        getRequest.storedFields("_none_")
        client.exists(getRequest, RequestOptions.DEFAULT)
    }

    /**
     * 删除数据
     * @param index
     * @param type
     * @param id
     * @return
     */
    def delete(String index, type, id) {
        DeleteRequest deleteRequest = new DeleteRequest(index, type, id)
        client.delete(deleteRequest, RequestOptions.DEFAULT)
    }

    /**
     * 搜索数据
     * @param index
     * @param query
     * @param size
     * @return
     */
    def search(String index, QueryBuilder query, int size = 10) {
        SearchRequest searchRequest = new SearchRequest(index)
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
        sourceBuilder.query(query)
        sourceBuilder.from(0)
        sourceBuilder.size(size)
        sourceBuilder.timeout(new TimeValue(1, TimeUnit.SECONDS))
        searchRequest.source(sourceBuilder)
       client.search(searchRequest, RequestOptions.DEFAULT)
    }

    /**
     * 滚动搜索
     * @param index
     * @param query
     * @param size
     */
    def searchScroll(String index, QueryBuilder query, int size = 10) {
        SearchRequest searchRequest = new SearchRequest(index)
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
        searchSourceBuilder.query(query)
        searchSourceBuilder.size(size)
        searchRequest.source(searchSourceBuilder)
        searchRequest.scroll(TimeValue.timeValueMinutes(1L))
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT)
        String scrollId = searchResponse.getScrollId()
        SearchHits hits = searchResponse.getHits()
        def searchHits = hits.getHits()
        while (searchHits != null && searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId)
            scrollRequest.scroll(TimeValue.timeValueMinutes(1L))
            searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT)
            scrollId = searchResponse.getScrollId()
            searchHits = searchResponse.getHits().getHits()
        }

    }

    def close() {
        client.close()
    }
}

初探 Client quot import index elasticsearch ide java软件测试

有关ES Client性能测试初探的更多相关文章

ruby-on-rails - 使用 Ruby on Rails 进行自动化测试 - 最佳实践 - 2
很好奇，就使用rubyonrails自动化单元测试而言，你们正在做什么？您是否创建了一个脚本来在cron中运行rake作业并将结果邮寄给您？git中的预提交Hook？只是手动调用？我完全理解测试，但想知道在错误发生之前捕获错误的最佳实践是什么。让我们理所当然地认为测试本身是完美无缺的，并且可以正常工作。下一步是什么以确保他们在正确的时间将可能有害的结果传达给您？最佳答案不确定您到底想听什么，但是有几个级别的自动代码库控制:在处理某项功能时，您可以使用类似autotest的内容获得关于哪些有效，哪些无效的即时反馈。要确保您的提
ruby - 使用 C 扩展开发 rubygem 时，如何使用 Rspec 在本地进行测试？ - 2
我正在编写一个包含C扩展的gem。通常当我写一个gem时，我会遵循TDD的过程，我会写一个失败的规范，然后处理代码直到它通过，等等......在“ext/mygem/mygem.c”中我的C扩展和在gemspec的“扩展”中配置的有效extconf.rb，如何运行我的规范并仍然加载我的C扩展？当我更改C代码时，我需要采取哪些步骤来重新编译代码？这可能是个愚蠢的问题，但是从我的gem的开发源代码树中输入“bundleinstall”不会构建任何native扩展。当我手动运行rubyext/mygem/extconf.rb时，我确实得到了一个Makefile(在整个项目的根目录中)，然后当
ruby - Ruby 的 Hash 在比较键时使用哪种相等性测试？ - 2
我有一个围绕一些对象的包装类，我想将这些对象用作散列中的键。包装对象和解包装对象应映射到相同的键。一个简单的例子是这样的:classAattr_reader:xdefinitialize(inner)@inner=innerenddefx;@inner.x;enddef==(other)@inner.x==other.xendenda=A.new(o)#oisjustanyobjectthatallowso.xb=A.new(o)h={a=>5}ph[a]#5ph[b]#nil,shouldbe5ph[o]#nil,shouldbe5我试过==、===、eq?并散列所有无济于事。
ruby - RSpec - 使用测试替身作为 block 参数 - 2
我有一些Ruby代码，如下所示:Something.createdo|x|x.foo=barend我想编写一个测试，它使用double代替block参数x，这样我就可以调用:x_double.should_receive(:foo).with("whatever").这可能吗？最佳答案 specify'something'dox=doublex.should_receive(:foo=).with("whatever")Something.should_receive(:create).and_yield(x)#callthere
ruby - Sinatra:运行 rspec 测试时记录噪音 - 2
Sinatra新手；我正在运行一些rspec测试，但在日志中收到了一堆不需要的噪音。如何消除日志中过多的噪音？我仔细检查了环境是否设置为:test，这意味着记录器级别应设置为WARN而不是DEBUG。spec_helper:require"./app"require"sinatra"require"rspec"require"rack/test"require"database_cleaner"require"factory_girl"set:environment,:testFactoryGirl.definition_file_paths=%w{./factories./test/
ruby-on-rails - 迷你测试错误 : "NameError: uninitialized constant" - 2
我遵循MichaelHartl的“RubyonRails教程:学习Web开发”，并创建了检查用户名和电子邮件长度有效性的测试(名称最多50个字符，电子邮件最多255个字符)。test/helpers/application_helper_test.rb的内容是:require'test_helper'classApplicationHelperTest在运行bundleexecraketest时，所有测试都通过了，但我看到以下消息在最后被标记为错误:ERROR["test_full_title_helper",ApplicationHelperTest,1.820016791]test
ruby - 即使失败也继续进行多主机测试 - 2
我已经构建了一些serverspec代码来在多个主机上运行一组测试。问题是当任何测试失败时，测试会在当前主机停止。即使测试失败，我也希望它继续在所有主机上运行。Rakefile:namespace:specdotask:all=>hosts.map{|h|'spec:'+h.split('.')[0]}hosts.eachdo|host|begindesc"Runserverspecto#{host}"RSpec::Core::RakeTask.new(host)do|t|ENV['TARGET_HOST']=hostt.pattern="spec/cfengine3/*_spec.r
ruby-on-rails - 如何使辅助方法在 Rails 集成测试中可用？ - 2
我在app/helpers/sessions_helper.rb中有一个帮助程序文件，其中包含一个方法my_preference，它返回当前登录用户的首选项。我想在集成测试中访问该方法。例如，这样我就可以在测试中使用getuser_path(my_preference)。在其他帖子中，我读到这可以通过在测试文件中包含requiresessions_helper来实现，但我仍然收到错误NameError:undefinedlocalvariableormethod'my_preference'.我做错了什么？require'test_helper'require'sessions_hel
ruby-on-rails - Cucumber 是否只是 rspec 的包装器以帮助将测试组织成功能？ - 2
只是想确保我理解了事情。据我目前收集到的信息，Cucumber只是一个“包装器”，或者是一种通过将事物分类为功能和步骤来组织测试的好方法，其中实际的单元测试处于步骤阶段。它允许您根据事物的工作方式组织您的测试。对吗？最佳答案有点。它是一种组织测试的方式，但不仅如此。它的行为就像最初的Rails集成测试一样，但更易于使用。这里最大的好处是您的session在整个Scenario中保持透明。关于Cucumber的另一件事是您(应该)从使用您的代码的浏览器或客户端的角度进行测试。如果您愿意，您可以使用步骤来构建对象和设置状态，但通常您
ruby-on-rails - 如何调试 cucumber 测试？ - 2
我有:When/^(?:|I)follow"([^"]*)"(?:within"([^"]*)")?$/do|link,selector|with_scope(selector)doclick_link(link)endend我打电话的地方:Background:GivenIamanexistingadminuserWhenIfollow"CLIENTS"我的HTML是这样的:CLIENTS我一直收到这个错误:.F-.F--U-----U(::)failedsteps(::)nolinkwithtitle,idortext'CLIENTS'found(Capybara::Element

ES Client性能测试初探

准备工作

学习资料

ES Client

HTTP请求

ES Client

ES Client 封装

测试用例

添加数据

ES Client API封装类

有关ES Client性能测试初探的更多相关文章

随机推荐