[elastic 8.x]java客户端连接elasticsearch与操作索引与文档

初始化客户端

引入相关依赖

<dependency><groupId>co.elastic.clients</groupId><artifactId>elasticsearch-java</artifactId><version>8.10.2</version>
</dependency>

初始化客户端

为了方便演示，我关闭了elasticsearch的安全验证，带安全验证的初始化方式将在最后专门介绍

String serverUrl="http://127.0.0.1:9200";
RestClient restClient=RestClient.builder(HttpHost.create(serverUrl)).build();
ElasticsearchTransport transport=new RestClientTransport(restClient,new JacksonJsonpMapper());
ElasticsearchClient esClient=new ElasticsearchClient(transport);

索引

创建索引

    void createIndex(){String mappings = "{\n" +"  \"properties\" : {\n" +"    \"id\" : {\n" +"      \"type\" : \"keyword\" \n" +"    },\n"+"    \"name\" : {\n" +"      \"type\" : \"text\",\n" +"      \"fields\" : {\n" +"        \"keyword\" : {\n" +"          \"type\" : \"keyword\",\n" +"          \"ignore_above\" : 256 \n" +"        }\n" +"      } \n" +"    }, \n" +"    \"price\" : { \n" +"      \"type\" : \"long\" \n" +"     } \n" +"  }\n" +"}\n";JsonpMapper mapper=esClient._transport().jsonpMapper();JsonParser parser= Json.createParser(new StringReader(mappings));CreateIndexRequest createIndexRequest=new CreateIndexRequest.Builder().index("test").mappings(TypeMapping._DESERIALIZER.deserialize(parser,mapper)).build();try {esClient.indices().create(createIndexRequest);} catch (IOException e) {throw new RuntimeException(e);}}

删除索引

    void deleteMapping(){try {DeleteIndexResponse response;response= esClient.indices().delete(deleteIndexRequest->deleteIndexRequest.index("test"));System.out.println(response.toString());} catch (IOException e) {throw new RuntimeException(e);}}

判断索引是否存在

    void existsIndex(){try {BooleanResponse hotel = esClient.indices().exists(existsIndexRequest -> existsIndexRequest.index("hotel"));System.out.println(hotel.value());} catch (IOException e) {throw new RuntimeException(e);}}

文档

新增文档

    void insertDoc(){Hotel hotel=hotelService.getById(61083L);HotelDoc hotelDoc=new HotelDoc(hotel);IndexRequest<HotelDoc> request=new IndexRequest.Builder<HotelDoc>().id("11").index("hotel").document(hotelDoc).build();try {IndexResponse index = esClient.index(request);System.out.println(index.id());} catch (IOException e) {throw new RuntimeException(e);}}

其中，HotelDoc是一个实体类

删除文档

    void deleteDoc(){try {esClient.delete(deleteRequest->deleteRequest.index("hotel").id("11"));} catch (IOException e) {throw new RuntimeException(e);}}

查询文档

    void searchDoc(){TermQuery termQuery= QueryBuilders.term().field("_id").value("11").build();SearchRequest request=new SearchRequest.Builder().index("hotel").query(termQuery._toQuery()).build();try {SearchResponse<HotelDoc> response=esClient.search(request,HotelDoc.class);//输出结果for(Hit<HotelDoc> hit:response.hits().hits()){System.out.println(hit.source());}} catch (IOException e) {throw new RuntimeException(e);}}

更新文档

    void updateDoc(){HotelDoc hotelDoc=new HotelDoc();//需要更新哪个字段就赋值哪个字段hotelDoc.setCity("xx");try {esClient.update(updateRequest->updateRequest.index("hotel").id("11").doc(hotelDoc),HotelDoc.class);} catch (IOException e) {throw new RuntimeException(e);}}

批量导入文档

    void insertMany(){List<Hotel> hotels=hotelService.list();List<HotelDoc> hotelDocs=hotels.stream().map(HotelDoc::new).collect(Collectors.toList());BulkRequest.Builder bl=new BulkRequest.Builder();for(HotelDoc hotelDoc:hotelDocs){bl.operations(op->op.index(idx->idx.index("hotel").id(hotelDoc.getId().toString()).document(hotelDoc)));}try {esClient.bulk(bl.refresh(Refresh.WaitFor).build());} catch (IOException e) {throw new RuntimeException(e);}}

连接Https集群

带安全验证的连接有点复杂，将下列代码中CA证书的位置改为实际所在的位置就行了。

通过用户名密码连接

password为elastic的密码，可以在我的另一篇文章中查看密码的重置方式
Docker安装部署[8.x]版本Elasticsearch+Kibana+IK分词器

    void makeConnection_https() throws CertificateException, IOException,NoSuchAlgorithmException, KeyStoreException, KeyManagementException {// 创建凭据提供器final CredentialsProvider credentialsProvider =new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY,new UsernamePasswordCredentials("elastic", password));// 设置CA证书路径Path caCertificatePath = Paths.get("E:\\tools\\elasticsearch-8.10.2\\config\\certs\\http_ca.crt");// 创建证书工厂CertificateFactory factory =CertificateFactory.getInstance("X.509");Certificate trustedCa;try (InputStream is = Files.newInputStream(caCertificatePath)) {// 从输入流中生成证书trustedCa = factory.generateCertificate(is);}// 创建密钥库KeyStore trustStore = KeyStore.getInstance("pkcs12");trustStore.load(null, null);trustStore.setCertificateEntry("ca", trustedCa);// 创建SSL上下文构建器SSLContextBuilder sslContextBuilder = SSLContexts.custom().loadTrustMaterial(trustStore, null);final SSLContext sslContext = sslContextBuilder.build();// 构建Rest客户端构建器RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "https")).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {return httpClientBuilder.setSSLContext(sslContext).setDefaultCredentialsProvider(credentialsProvider);}});// 构建Rest客户端RestClient restClient = builder.build();// 创建Rest客户端传输ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());esClient = new ElasticsearchClient(transport);
//        asyncClient = new ElasticsearchAsyncClient(transport);}

通过ApiKey连接

ApiKey在Kibana的Security下生成

     void makeConnection_token() throws CertificateException, IOException,NoSuchAlgorithmException, KeyStoreException, KeyManagementException {// 定义CA证书路径Path caCertificatePath = Paths.get("E:\\tools\\elasticsearch-8.10.2\\config\\certs\\http_ca.crt");// 创建X.509证书工厂CertificateFactory factory =CertificateFactory.getInstance("X.509");Certificate trustedCa;try (InputStream is = Files.newInputStream(caCertificatePath)) {// 从输入流中生成X.509证书trustedCa = factory.generateCertificate(is);}// 创建PKCS12密钥库KeyStore trustStore = KeyStore.getInstance("pkcs12");trustStore.load(null, null);// 将CA证书添加到密钥库trustStore.setCertificateEntry("ca", trustedCa);// 创建SSL上下文构建器，并设置信任材料SSLContextBuilder sslContextBuilder = SSLContexts.custom().loadTrustMaterial(trustStore, null);final SSLContext sslContext = sslContextBuilder.build();// 创建Rest客户端构建器RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "https")).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {return httpClientBuilder.setSSLContext(sslContext);}});// 设置默认请求头Header[] defaultHeaders =new Header[]{new BasicHeader("Authorization","ApiKey yourApiKey")};builder.setDefaultHeaders(defaultHeaders);// 构建Rest客户端RestClient restClient = builder.build();// 创建基于RestClient的传输方式ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());// 创建Elasticsearch客户端esClient = new ElasticsearchClient(transport);}