快速开始高性能 Elasticsearch 客户端 bboss

作者：大河

2023-03-24
湖南
本文字数：10901 字
阅读完需：约 36 分钟

Quick Start

Elasticsearch/Opensearch Bboss--高性能 Elasticsearch Java RestClient

主要特点：学习成本低，上手快，代码简洁，安全高效，客户端负载容灾，兼容性好，易于集成

A highlevel http rest client.
A high performence o/r mapping rest client.
A dsl and sql rest client.
Support Elasticsearch 1.x,2.x,5.x,6.x,7.x,8.x,+
Support Opensearch 1.x,2.x,+
Support Spring boot 1.x,2.x,3.x
可在普通 maven 项目和其他 java 工程中集成 bboss
操作返回的结果可以是原始 json 报文、PO 对象、List 集合、Map 对象以及分页查询、聚合查询、高亮检索封装对象，可以方便的从结果中获取索引文档 id、score 等元数据信息
学习成本低
1）无需任何配置即可完成增删改、简单查询操作，复杂的查询才需编写和配置 dsl；
2）通过 bboss ClientInterface 接口即可完成所有的 Elasticsearch 操作，只需学习 Elasticsearch 官方 dsl 语言，免除额外工具 api 学习成本，无需引入其他充血、ActiveRecord 概念模型；
3）可以借助 kibana devtool 调试 dsl，调试通过后直接放入 bboss dsl 配置文件，调整放置检索变量参数即可完成各种复杂的 Elasticsearch 查询检索操作
4）bboss 完全支持和兼容 Elasticsearch 各个版本的 dsl 语法，基于 bboss 对接各个版本的 Elasticsearch，不会对 Elasticsearch 造成任何的功能损耗，可以说 Elasticsearch 的既是 bboss 的，免除 Elasticsearch 版本升级的后顾之忧。

1.快速集成和应用 Bboss

1.1 maven 项目集成

快速集成，导入 BBoss maven 坐标:

    <dependency>        <groupId>com.bbossgroups.plugins</groupId>        <artifactId>bboss-elasticsearch-rest-jdbc</artifactId>        <version>6.8.7</version>    </dependency>

复制代码

如果是 spring boot 项目，还需导入以下 maven 坐标:

    <dependency>        <groupId>com.bbossgroups.plugins</groupId>        <artifactId>bboss-elasticsearch-spring-boot-starter</artifactId>        <version>6.8.7</version>    </dependency>

复制代码

1.2 一般 java 项目集成

一般的 java 项目集成，直接从百度网盘下载 bboss 解压，将 bboss jar 包以及依赖 jar 文件导入 java 工程即可

链接：bboss 6.8.7提取码：nw76

1.3 快速配置

快速配置，在 application.properties 文件中增加 Elasticsearch 服务器地址和认证口令(可选)配置即可

#elasticsearch.rest.hostNames=10.180.211.27:9200,10.180.211.28:9200,10.180.211.29:9200 
elasticsearch.rest.hostNames=10.21.20.168:9200
#x-pack or searchguard security authentication and password configuration
elasticUser=elasticelasticPassword=changeme

复制代码

spring boot 配置

spring.elasticsearch.bboss.elasticsearch.rest.hostNames=10.180.211.27:9200#spring.elasticsearch.bboss.elasticsearch.rest.hostNames=10.180.211.27:9200,10.180.211.28:9200,10.180.211.29:9200 
##support x-pack and searchguard
spring.elasticsearch.bboss.elasticUser=elastic
spring.elasticsearch.bboss.elasticPassword=changeme

复制代码

1.4 快速操作和访问 Elasticsearch

一行代码插入/修改

// 添加 / 修改文档，如果文档 id 存在则修改，不存在则插入

clientUtil.addDocument("agentinfo",//索引名称                agentInfo);//需添加/修改的索引数据对象

复制代码

一行代码批量插入/修改文档

// 添加 / 修改文档，如果文档 id 存在则修改，不存在则插入

List<AgentInfo> agentInfos = ....;clientUtil.addDocuments("agentinfo",//索引名称                agentInfos);//需批量添加/修改的索引数据对象集合

复制代码

一行代码分页 / 高亮检索

ESDatas<TAgentInfo> data //ESDatas为查询结果集对象，封装了返回的当前查询的List<TAgentInfo>结果集和符合条件的总记录数totalSize            = clientUtil.searchList("trace-*/_search",//查询操作，查询indices trace-*中符合条件的数据                                "queryServiceByCondition",//通过名称引用配置文件中的query dsl语句                                traceExtraCriteria,//查询条件封装对象                                TAgentInfo.class);//指定返回的po对象类型，po对象中的属性与indices表中的文档filed名称保持一致//获取当前页结果对象列表        List<TAgentInfo> demos = data.getDatas();        //获取总记录数        long totalSize = data.getTotalSize();

复制代码

根据文档 id 获取文档

Demo demo = clientUtil.getDocument("demo",//索引表      "2",//文档id      Demo.class);//指定返回对象类型

复制代码

根据字段直接获取文档

String document = clientInterface.getDocumentByField("demo",//索引名称                  "applicationName.keyword",//字段名称                  "blackcatdemo2");//字段值 Map document = clientInterface.getDocumentByField("demo","applicationName.keyword","blackcatdemo2",Map.class); DemoPo document = clientInterface.getDocumentByField("demo","applicationName.keyword","blackcatdemo2",DemoPo.class);

复制代码

一行代码根据字段值进行分页查找

ESDatas<Map> documents = clientInterface.searchListByField("demo",//索引名名称                                          "applicationName.keyword", //检索字段名称                                          "blackcatdemo2",//检索值                                           Map.class,  //返回结果类型，可以是po对象类型也可以是map类型                                           0,  //分页起始位置                                           10); //分页每页记录数//获取当前页结果对象列表        List<Map> demos = data.getDatas();        //获取匹配的总记录数        long totalSize = data.getTotalSize();

复制代码

一行代码删除文档

clientUtil.deleteDocument("demo",//索引表          "2");//文档id

复制代码

一行代码批量删除文档

//批量删除文档        clientUtil.deleteDocuments("demo",//索引表                new String[]{"2","3"});//批量删除文档ids

复制代码

api 方法可以指定特定的Elasticsearch 集群进行操作

所有 api 可以直接指定数据源操作，指哪打哪，下面展示了在一段代码里面同时操作两个 Elasticsearch 集群功能：datasourceName1 和 datasourceName2

ESDatas<Demo> esDatas1 =             clientUtil.searchListWithCluster(datasourceName1,//指定操作的Elasticsearch集群数据源名称                  "demo1/_search",//demo为索引表，_search为检索操作action            "searchDatas",//esmapper/demo7.xml中定义的dsl语句            params,//变量参数            Demo.class);//返回的文档封装对象类型
ESDatas<Demo> esDatas2 =             clientUtil.searchListWithCluster(datasourceName2,//指定操作的Elasticsearch集群数据源名称                  "demo2/_search",//demo为索引表，_search为检索操作action            "searchDatas",//esmapper/demo7.xml中定义的dsl语句            params,//变量参数            Demo.class);//返回的文档封装对象类型

复制代码

datasourceName1 和 datasourceName2 可是两个相同版本的 Elasticsearch，亦可以是两个不同版本的 Elasticsearch，对 Elasticsearch 兼容性非常棒。

简单实用的异步批处理器

通过简单实用的异步批处理器，可以大大提升 Elasticsearch 数据写入处理的性能和吞吐量

参考文档：https://esdoc.bbossgroups.com/#/bulkProcessor

2.客户端组件 ClientInterface 实例获取

一般项目通过 ElasticSearchHelper 获取 elasticsearch rest client api 实例：

//创建加载配置文件的客户端实例，单实例多线程安全ClientInterface clientUtil = ElasticSearchHelper.getConfigRestClientUtil("esmapper/demo.xml");//创建直接操作dsl的客户端实例，单实例多线程安全ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil() ;

复制代码

Spring boot 项目使用 BBossESStarter 获取 elasticsearch rest client api 实例:

@Autowiredprivate BBossESStarter bbossESStarter;//Create a client tool to load configuration files, single instance multithreaded securityClientInterface clientUtil = bbossESStarter.getConfigRestClient("esmapper/demo.xml");    //Build a create/modify/get/delete document client object, single instance multi-thread security    ClientInterface clientUtil = bbossESStarter.getRestClient();

复制代码

Elasticsearch bboss 开发指南:

https://esdoc.bbossgroups.com/#/document-crud

https://esdoc.bbossgroups.com/#/development

3.配置和使用-进阶

在项目 resources 目录下修改 application.properties 文件（如果不存在则新建 application.properties 文件），根据项目类型做添加相应配置:

3.1 普通 java 和 maven 项目配置

#Cluster addresses are separated by commas
#elasticsearch.rest.hostNames=10.180.211.27:9200,10.180.211.28:9200,10.180.211.29:9200
elasticsearch.rest.hostNames=10.21.20.168:9200

复制代码

如需启用HTTPS协议, 则在 elasticsearch 地址前面添加 https 协议头，并设置 useHttps 属性为 true:

elasticsearch.useHttps=trueelasticsearch.rest.hostNames=https://10.180.211.27:9280,https://10.180.211.27:9281,https://10.180.211.27:9282

复制代码

如果启用了 x-pack or searchguard 安全认证机制, 则还需在 application.properties 文件配置认证账号和口令:

#x-pack or searchguard security authentication and password configurationelasticUser=elasticelasticPassword=changeme

复制代码

接下来就可以编写 java 代码验证集成是否成功:

import org.frameworkset.elasticsearch.ElasticSearchHelper;import org.frameworkset.elasticsearch.client.ClientInterface;import org.frameworkset.elasticsearch.entity.ESDatas;import org.frameworkset.elasticsearch.scroll.HandlerInfo;import org.frameworkset.elasticsearch.scroll.ScrollHandler;import org.junit.Test;
import java.util.List;import java.util.Map;
public class TestInit {   @Test   public void init(){
      ClientInterface clientUtil = ElasticSearchHelper.getRestClientUtil();      //get elasticsearch cluster state      String result = clientUtil.executeHttp("_cluster/state?pretty",ClientInterface.HTTP_GET);          //check indice twitter and index type tweet exist or not.      //适用于elasticsearch 6及以下版本有类型，7以上版本无类型      boolean exist1 = clientUtil.existIndiceType("twitter","tweet");      System.out.println("twitter  tweet type exist:"+exist1);      //适用于 Elasticsearch7以上的版本，check indice twitter exist or not，      exist1 = clientUtil.existIndice("twitter");      System.out.println("twitter exist:"+exist1);      //count documents in indice twitter      long count = clientUtil.countAll("twitter");      System.out.println(count);          //Get All documents of indice twitter,DEFAULT_FETCHSIZE is 5000      //返回对象类型为Map，也可以指定为特定的PO对象，适用于记录数量不大的表      ESDatas<Map> esDatas = clientUtil.searchAll("twitter", Map.class);      //从esDatas中获取检索到的记录集合      List<Map> datas = esDatas.getDatas();      //从esDatas中获取检索到的记录总数      long totalSize = esDatas.getTotalSize();          //Get All documents of indice twitter,Set fetchsize to 10000, Using ScrollHandler to process each batch of datas.      //指定批处理器分批次处理数据，适用于记录数量比较大的全表表数据查询      clientUtil.searchAll("twitter",10000,new ScrollHandler<Map>() {         public void handle(ESDatas<Map> esDatas, HandlerInfo handlerInfo) throws Exception {            List<Map> dataList = esDatas.getDatas();            System.out.println("TotalSize:"+esDatas.getTotalSize());            if(dataList != null) {               System.out.println("dataList.size:" + dataList.size());            }            else            {               System.out.println("dataList.size:0");            }            //do something other such as do a db query.            //SQLExecutor.queryList(Map.class,"select * from td_sm_user");         }      },Map.class);      //Use slice parallel scoll query all documents of indice  twitter by 2 thread tasks. DEFAULT_FETCHSIZE is 5000      //You can also use ScrollHandler to process each batch of datas on your own.      esDatas = clientUtil.searchAllParallel("twitter", Map.class,2);      //指定批处理器分批次处理数据（适用于数据量比较大的表），并行检索和处理表数据源，线程数量为2，      clientUtil.searchAllParallel("twitter",10000,new ScrollHandler<Map>() {         public void handle(ESDatas<Map> esDatas, HandlerInfo handlerInfo) throws Exception {            List<Map> dataList = esDatas.getDatas();            System.out.println("TotalSize:"+esDatas.getTotalSize());            if(dataList != null) {               System.out.println("dataList.size:" + dataList.size());            }            else            {               System.out.println("dataList.size:0");            }            //do something other such as do a db query.            //SQLExecutor.queryList(Map.class,"select * from td_sm_user");         }      },Map.class,2);   }}

复制代码

maven project 案例下载地址:

https://gitee.com/bboss/eshelloword-booter

参考文档

集成和配置 Elasticsearch bboss：

https://esdoc.bbossgroups.com/#/common-project-with-bboss

3.2 spring boot maven 项目配置

spring.elasticsearch.bboss.elasticsearch.rest.hostNames=127.0.0.1:9200

复制代码

如需启用HTTPS协议, 则在 elasticsearch 地址前面添加 https 协议头，并设置 useHttps 属性为 true:

spring.elasticsearch.bboss.elasticsearch.useHttps=truespring.elasticsearch.bboss.elasticsearch.rest.hostNames=https://10.180.211.27:9280,https://10.180.211.27:9281,https://10.180.211.27:9282

复制代码

如果启用了 x-pack or searchguard 安全认证机制, 则还需在 application.properties 文件配置认证账号和口令:

##support x-pack and searchguardspring.elasticsearch.bboss.elasticUser=elasticspring.elasticsearch.bboss.elasticPassword=changeme

复制代码

Spring boot 需要使用注解 @Autowired 注入一个 BBossESStarter 对象来获取 elasticsearch rest client api 实例对象（每次获取到的实例是单实例多线程安全的）：

@Autowiredprivate BBossESStarter bbossESStarter;//Create a client tool to load configuration files, single instance multithreaded securityClientInterface configClientUtil = bbossESStarter.getConfigRestClient(mappath);    //Build a create/modify/get/delete document client object, single instance multi-thread securityClientInterface clientUtil = bbossESStarter.getRestClient();

复制代码

Spring boot 示例代码:

The dsl xml file this example used esmapper/demo.xml

package com.example.esbboss.service;
import com.example.esbboss.entity.Demo;import com.example.esbboss.entity.DemoSearchResult;import org.frameworkset.elasticsearch.ElasticSearchException;import org.frameworkset.elasticsearch.boot.BBossESStarter;import org.frameworkset.elasticsearch.client.ClientInterface;import org.frameworkset.elasticsearch.entity.ESDatas;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;
import java.text.DateFormat;import java.text.ParseException;import java.text.SimpleDateFormat;import java.util.Date;import java.util.HashMap;import java.util.List;import java.util.Map;

@Servicepublic class DocumentCRUD7 {   private Logger logger = LoggerFactory.getLogger(DocumentCRUD7.class);   @Autowired   private BBossESStarter bbossESStarter;   //DSL config file path   private String mappath = "esmapper/demo7.xml";

   public void dropAndCreateAndGetIndice(){      //Create a client tool to load configuration files, single instance multithreaded security      ClientInterface clientUtil = bbossESStarter.getConfigRestClient(mappath);      try {         //To determine whether the indice demo exists, it returns true if it exists and false if it does not         boolean exist = clientUtil.existIndice("demo");
         //Delete mapping if the indice demo already exists         if(exist) {            String r = clientUtil.dropIndice("demo");            logger.debug("clientUtil.dropIndice(\"demo\") response:"+r);             }         //Create index demo         clientUtil.createIndiceMapping("demo",//The indice name               "createDemoIndice");//Index mapping DSL script name, defined createDemoIndice in esmapper/demo.xml         String demoIndice = clientUtil.getIndice("demo");//Gets the newly created indice structure         logger.info("after createIndiceMapping clientUtil.getIndice(\"demo\") response:"+demoIndice);      } catch (ElasticSearchException e) {         // TODO Auto-generated catch block         e.printStackTrace();      }
   }


   public void addAndUpdateDocument()  {      //Build a create/modify/get/delete document client object, single instance multi-thread security      ClientInterface clientUtil = bbossESStarter.getRestClient();      //Build an object as index document      Demo demo = new Demo();      demo.setDemoId(2l);//Specify the document id, the unique identity, and mark with the @ESId annotation. If the demoId already exists, modify the document; otherwise, add the document      demo.setAgentStarttime(new Date());      demo.setAgentStarttimezh(new Date());      demo.setApplicationName("blackcatdemo2");      demo.setContentbody("this is content body2");      demo.setName("liudehua");      demo.setOrderId("NFZF15045871807281445364228");      demo.setContrastStatus(2);

      //Add the document and force refresh      String response = clientUtil.addDocument("demo",//indice name            demo,"refresh=true");


      logger.debug("Print the result：addDocument-------------------------");      logger.debug(response);          demo = new Demo();      demo.setDemoId(3l);//Specify the document id, the unique identity, and mark with the @ESId annotation. If the demoId already exists, modify the document; otherwise, add the document      demo.setAgentStarttime(new Date());      demo.setApplicationName("blackcatdemo3");      demo.setContentbody("this is content body3");      demo.setName("zhangxueyou");      demo.setOrderId("NFZF15045871807281445364228");      demo.setContrastStatus(3);      demo.setAgentStarttime(new Date());      demo.setAgentStarttimezh(new Date());          //Add the document and force refresh      response = clientUtil.addDocument("demo",//indice name            demo,"refresh=true");          //Get the document object according to the document id, and return the Demo object      demo = clientUtil.getDocument("demo",//indice name            "2",//document id            Demo.class);          //update document      demo = new Demo();      demo.setDemoId(2l);//Specify the document id, the unique identity, and mark with the @ESId annotation. If the demoId already exists, modify the document; otherwise, add the document      demo.setAgentStarttime(new Date());      demo.setApplicationName("blackcatdemo2");      demo.setContentbody("this is modify content body2");      demo.setName("刘德华modify\t");      demo.setOrderId("NFZF15045871807281445364228");      demo.setContrastStatus(2);      demo.setAgentStarttimezh(new Date());      //Execute update and force refresh      response = clientUtil.addDocument("demo",//index name            demo,"refresh=true");

      //Get the modified document object according to the document id and return the json message string      response = clientUtil.getDocument("demo",//indice name            "2");//document id      logger.debug("Print the modified result:getDocument-------------------------");      logger.debug(response);



      logger.debug("Print the modified result：getDocument-------------------------");      logger.debug(response);

   }
   public void deleteDocuments(){      //Build a create/modify/get/delete document client object, single instance multi-thread security      ClientInterface clientUtil = bbossESStarter.getRestClient();      //Batch delete documents      clientUtil.deleteDocuments("demo",//indice name            new String[]{"2","3"});//Batch delete document ids   }
   /**    * Use slice parallel scoll query all documents of indice demo by 2 thread tasks. DEFAULT_FETCHSIZE is 5000    */   public void searchAllPararrel(){      ClientInterface clientUtil = bbossESStarter.getRestClient();      ESDatas<Demo> esDatas = clientUtil.searchAllParallel("demo", Demo.class,2);   }


   /**    * Search the documents    */   public DemoSearchResult search()   {      //Create a load DSL file client instance to retrieve documents, single instance multithread security      ClientInterface clientUtil = bbossESStarter.getConfigRestClient(mappath);      //Set query conditions, pass variable parameter values via map,key for variable names in DSL      //There are four variables in the DSL:      //        applicationName1      //        applicationName2      //        startTime      //        endTime      Map<String,Object> params = new HashMap<String,Object>();      //Set the values of applicationName1 and applicationName2 variables      params.put("applicationName1","blackcatdemo2");      params.put("applicationName2","blackcatdemo3");      DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");      //Set the time range, and accept the long value as the time parameter      try {         params.put("startTime",dateFormat.parse("2017-09-02 00:00:00").getTime());      } catch (ParseException e) {         e.printStackTrace();      }      params.put("endTime",new Date().getTime());

      //Execute the query      ESDatas<Demo> esDatas =  //ESDatas contains a collection of currently retrieved records, up to 1000 records, specified by the size attribute in the DSL            clientUtil.searchList("demo/_search",//demo as the indice, _search as the search action            "searchDatas",//DSL statement name defined in esmapper/demo.xml            params,//Query parameters            Demo.class);//Data object type Demo returned      //Gets a list of result objects and returns max up to 1000 records (specified in DSL)      List<Demo> demos = esDatas.getDatas();
//    String json = clientUtil.executeRequest("demo/_search",//demo as the index table, _search as the search action//          "searchDatas",//DSL statement name defined in esmapper/demo.xml//          params);//Query parameters//    String json = com.frameworkset.util.SimpleStringUtil.object2json(demos);      //Gets the total number of records      long totalSize = esDatas.getTotalSize();      DemoSearchResult demoSearchResult = new DemoSearchResult();      demoSearchResult.setDemos(demos);      demoSearchResult.setTotalSize(totalSize);      return demoSearchResult;   }

}