java实现es操作

bjtang大约 4 分钟

gradle项目

#你需要在项目依赖中引入以下依赖项
dependencies {
    implementation 'co.elastic.clients:elasticsearch-java:8.0.1'
    implementation 'jakarta.json:jakarta.json-api:2.0.1'
    implementation 'com.alibaba:fastjson:1.2.83' 
}

maven项目请在此网站查阅依赖
https://central.sonatype.com/artifact/org.apache.httpcomponents/httpclient/versions
普通项目请下载相关依赖手动引入

在使用JAVA进行操作es时,你应该提前熟悉es的基本操作
参考大数据-示例集合-elasticsearch

初步使用

让我们从进行一次基本的查询开始

//引入相关依赖
//EsCRUD.java
import java.io.IOException;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import co.elastic.clients.json.JsonData;

public class EsCRUD {
 public static void main(String[] args) throws ElasticsearchException, IOException {
  // 设置账号密码
  final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  credentialsProvider.setCredentials(AuthScope.ANY,
    new UsernamePasswordCredentials("elastic", "nsQywcgUoC3ZljMJJ0p1"));
  // 配置连接信息
  RestClientBuilder builder = RestClient.builder(new HttpHost("192.168.1.74", 9200))
    .setHttpClientConfigCallback(new HttpClientConfigCallback() {
     @Override
     public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
      return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
     }
    });
  RestClient restClient = builder.build();
  // 使用Jackson映射器创建传输层
  RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
  // 创建客户端
  ElasticsearchClient client = new ElasticsearchClient(transport);
  // 进行查询name为TTT的数据
  SearchResponse<JsonData> search = client.search(
    s -> s.index("test_index2").query(q -> q.match(m -> m.field("name").query("TTT"))), JsonData.class);

  // 检查 hits 是否为空
  if (search.hits().hits().isEmpty()) {
   System.err.println("No hits found");
  }

  for (Hit<JsonData> hit : search.hits().hits()) {
   System.err.println("Processing hit...");
   JsonData source = hit.source();
   if (source != null) {
    System.err.println(source);
    //System.err.println("Source: " + source.getId()+source.getName()+source.getCreateAt());
   } else {
    System.err.println("Source is null");
   }
  }
  System.err.println("执行完毕");
  // 关闭客户端
  try {
   transport.close();
   restClient.close();
  } catch (IOException e) {
  e.printStackTrace();
  }
 }
}

#输出结果
Processing hit...
{"id":"2016113136","name":"TTT","createAt":"1999-01-01"}
执行完毕

首先我们进行es的连接,然后进行查询,并将查询到的JSONData数据输出到控制台那么我们进行标准化改造一下,使之更加通用

//新建实体类Test_index2.java
public class Test_index2 {
 private Long id;
 private String name;
 private String createAt;
 public Long getId() {
  return id;
 }
 public void setId(Long id) {
  this.id = id;
 }
 public String getName() {
  return name;
 }
 public void setName(String name) {
  this.name = name;
 }
 public String getCreateAt() {
  return createAt;
 }
 public void setCreateAt(String createAt) {
  this.createAt = createAt;
 }
}

//查询时绑定实体类
  SearchResponse<Test_index2> search = client.search(
    s -> s.index("test_index2").query(q -> q.match(m -> m.field("name").query("TTT"))), Test_index2.class);
//结果操作时使用实体类进行操作
   for (Hit<Test_index2> hit : search.hits().hits()) {
   System.err.println("Processing hit...");
   Test_index2 source = hit.source();
   if (source != null) {
   // System.err.println(source);
    System.err.println("Source: " + source.getId()+source.getName()+source.getCreateAt());
   } else {
    System.err.println("Source is null");
   }
  }

//单独给RestClient提出来方便调用
public static RestClient getRestClient() {
  // 设置账号密码
  final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  credentialsProvider.setCredentials(AuthScope.ANY,
    new UsernamePasswordCredentials("elastic", "nsQywcgUoC3ZljMJJ0p1"));

  RestClientBuilder builder = RestClient.builder(new HttpHost("192.168.1.74", 9200))
    .setHttpClientConfigCallback(new HttpClientConfigCallback() {
     @Override
     public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
      return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
     }
    });
  RestClient restClient = builder.build();
  return restClient;
 }

#输出结果
Processing hit...
Source: 2016113136TTT1999-01-01
执行完毕

我们已经完成es的基础操作了!接下来我们进一步了解es的具体操作

索引

创建索引

 public static void addIndex() throws ElasticsearchException, IOException {
  // 创建连接
  RestClient restClient = getRestClient();
  RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
  ElasticsearchClient client = new ElasticsearchClient(transport);
  // 创建索引
  CreateIndexResponse createIndexResponse = client.indices().create(c -> c.index("newapi"));
  // 打印结果
  System.out.println(createIndexResponse.acknowledged());
  // 关闭客户端
  try {
   transport.close();
   restClient.close();
  } catch (IOException e) {
   e.printStackTrace();
  }
 }

#输出结果
true

查询索引

 public static void getIndex() throws ElasticsearchException, IOException {
  // 创建连接
  RestClient restClient = getRestClient();
  RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
  ElasticsearchClient client = new ElasticsearchClient(transport);
  // 查询索引
  GetIndexResponse createIndexResponse = client.indices().get(e->e.index("test_index2"));
  // 打印结果
  
  System.err.println(String.join(",", createIndexResponse.result().keySet()));
  // 关闭客户端
  try {
   transport.close();
   restClient.close();
  } catch (IOException e) {
   e.printStackTrace();
  }
 }

#输出结果
test_index2_v2

删除索引

 public static void delIndex() throws ElasticsearchException, IOException {
  RestClient restClient = getRestClient();
  RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
  ElasticsearchClient client = new ElasticsearchClient(transport);
  // 删除索引
  DeleteIndexResponse deleteIndexResponse = client.indices().delete(e->e.index("newapi"));
  System.out.println(deleteIndexResponse.acknowledged());
  // 关闭客户端
  try {
   transport.close();
   restClient.close();
  } catch (IOException e) {
   e.printStackTrace();
  }
 }

#输出结果
true

文档

添加文档

 public static void addDoc() throws ElasticsearchException, IOException {
//...省略相同构造
  Test_index2 test = new Test_index2();
  test.setName("添加测试");
  test.setId(2016113205L);
  test.setCreateAt("2000-01-09");
  // 构建一个创建Doc的请求
  CreateResponse createResponse = client.create(e->e.index("test_index2").id(test.getId().toString()).document(test));
  System.err.println(createResponse.result());
  // ...省略关闭客户端
 }

#输出结果
Created
#如果在java请求中缺少id,不会自动生成id而是报错

修改文档

 public static void updateDoc() throws ElasticsearchException, IOException {
  //...省略相同构造
 Test_index2 test = new Test_index2();
  test.setName("修改测试");
//  test.setId(2016113205L);
//  test.setCreateAt("2000-01-09");
  // 构建修改文档的请求
  UpdateResponse<Test_index2> response = client.update(e -> e.index("test_index2").id("2016113205").doc(test), Test_index2.class);
  System.err.println(response.result());
    // ...省略关闭客户端
  }

#输出结果
Updated
#在这里注释了test.setId,在es中数据id字段不受影响

查询文档

 public static void getDoc() throws ElasticsearchException, IOException {
  //...省略相同构造
  GetResponse<Test_index2> response = client.get(e -> e.index("test_index2").id("2016113205"), Test_index2.class);
  System.err.println(response.source().getName()+response.source().getId()+response.source().getCreateAt());
  // ...省略关闭客户端
}

#输出结果
修改测试20161132052000-01-09

删除文档

public static void delDoc() throws ElasticsearchException, IOException {
  //...省略相同构造
  DeleteResponse delete = client.delete(e -> e.index("test_index2").id("2016113205"));
  System.err.println(delete.result());
  // ...省略关闭客户端
}

#输出结果
Deleted

批量添加文档

 public static void addListDoc() throws ElasticsearchException, IOException {
  //...省略相同构造
  List<BulkOperation> bulkOperations = new ArrayList<>();

  List<Test_index2> dataList = new ArrayList<>();
  Test_index2 data1=new Test_index2();
  data1.setId(2013113223L);
  data1.setName("批量插入1");
  data1.setCreateAt("1999-01-05");
  
  Test_index2 data2=new Test_index2();
  data2.setId(2013113224L);
  data2.setName("批量插入2");
  data2.setCreateAt("1999-01-06");
  
  dataList.add(data1);
  dataList.add(data2);
      for (Test_index2 data : dataList) {
             bulkOperations.add(new BulkOperation.Builder()
                 .index(idx -> idx
                     .index("test_index2")
                     .id(data.getId().toString())
                     .document(data)
                 )
                 .build()
             );
         }
  BulkResponse response = client.bulk(e -> e.index("test_index2").operations(bulkOperations));
  System.err.println(response.took());
   response.items().forEach(item -> System.err.println(item.result()));
  // ...省略关闭客户端

#输出结果
1
created
created