当前位置:首页 > 编程教程 > java技术文章 > java使用es查询的示例代码

java使用es查询的用法实例

  • 发布时间:
  • 作者:码农之家
  • 点击:312

这篇文章主要知识点是关于java、es、查询、实例、使用java操作elasticsearch的具体方法 的内容,如果大家想对相关知识点有系统深入的学习,可以参阅以下电子书

数据结构与抽象:Java语言描述
  • 类型:java大小:131.8 MB格式:PDF作者:弗兰克M.卡拉诺
立即下载
JavaScript启示录
JavaScript启示录原书完整版
  • 类型:JS大小:24.4 MB格式:PDF作者:Cody
立即下载

java使用es查询的示例代码

众所周知,elasticsearch简称es,它是基于基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎,基于RESTful web接口。Elasticsearch是用Java开发的,并作为Apache许可条款下的开放源码发布,是当前流行的企业级搜索引擎。设计用于云计算中,能够达到实时搜索,稳定,可靠,快速,安装使用方便。

我们建立一个网站或应用程序,并要添加搜索功能,但是想要完成搜索工作的创建是非常困难的。我们希望搜索解决方案要运行速度快,我们希望能有一个零配置和一个完全免费的搜索模式,我们希望能够简单地使用JSON通过HTTP来索引数据,我们希望我们的搜索服务器始终可用,我们希望能够从一台开始并扩展到数百台,我们要实时搜索,我们要简单的多租户,我们希望建立一个云的解决方案。因此我们利用Elasticsearch来解决所有这些问题以及可能出现的更多其它问题。

在java中使用es时,无非想解决的是查询速度不够快,效率不够高问题,单一从数据库里查询数据已经不能拿满足当前的业务需求,ok!那么现在我们来讲述一下如何在java中使用到es这个神奇的搜索服务器呢,首先,你得要去引用es的依赖包,依赖如下:

<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>5.5.0</version>
  </dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>5.5.0</version>
  </dependency>
     <dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>5.5.0</version>
 </dependency>

好的,添加完依赖之后,进入到es使用阶段,对了,有个问题得需要说明一下,要使用es的话需要安装jdk1.8工具包并且tomcat最好是7.05以上的版本,不然es会不支持,集成到项目里面可能会报错!,还有es的安装和下载,这里附下载地址下载地址:https://www.elastic.co/downloads/elasticsearch

一切准备就绪后接下来进入到我们真正期待的时刻,什么呢,没错,集成es,究竟如何在java里面搜索查询es服务器里面的东西呢,让我一一来为你揭晓:

首先我们建议一个es的工具类

package com.osa.utils;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.URL;
import java.net.URLConnection;
import java.net.URLEncoder;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.http.client.ClientProtocolException;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;

public class HTTPSentUtils {
/**
  * 向指定URL发送GET方法的请求
  * 
  * @param url
  *   发送请求的URL
  * @param param
  *   请求参数,请求参数应该是 name1=value1&name2=value2 的形式。
  * @return URL 所代表远程资源的响应结果
  */
 public static String sendGet(String url, String param) {
  String result = "";
  BufferedReader in = null;
  try {
   String urlNameString = url + "?" + param;
   URL realUrl = new URL(urlNameString);
   // 打开和URL之间的连接
   URLConnection connection = realUrl.openConnection();
   // 设置通用的请求属性
   connection.setRequestProperty("accept", "*/*");
   connection.setRequestProperty("connection", "Keep-Alive");
   connection.setRequestProperty("user-agent",
     "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
   // 建立实际的连接
   connection.connect();
   // 获取所有响应头字段
   Map<String, List<String>> map = connection.getHeaderFields();
   // 遍历所有的响应头字段
   for (String key : map.keySet()) {
    System.out.println(key + "--->" + map.get(key));
   }
   // 定义 BufferedReader输入流来读取URL的响应
   in = new BufferedReader(new InputStreamReader(
     connection.getInputStream()));
   String line;
   while ((line = in.readLine()) != null) {
    result += line;
   }
  } catch (Exception e) {
   System.out.println("发送GET请求出现异常!" + e);
   e.printStackTrace();
  }
  // 使用finally块来关闭输入流
  finally {
   try {
    if (in != null) {
     in.close();
    }
   } catch (Exception e2) {
    e2.printStackTrace();
   }
  }
  return result;
 }
 /**
  * 向指定 URL 发送POST方法的请求
  * 
  * @param url
  *   发送请求的 URL
  * @param param
  *   请求参数,请求参数应该是 name1=value1&name2=value2 的形式。
  * @return 所代表远程资源的响应结果
  */
 public static String sendPost(String url, String param) {
  PrintWriter out = null;
  BufferedReader in = null;
  String result = "";
  try {
   URL realUrl = new URL(url);
   // 打开和URL之间的连接
   URLConnection conn = realUrl.openConnection();
   // 设置通用的请求属性
   conn.setRequestProperty("accept", "*/*");
   conn.setRequestProperty("connection", "Keep-Alive");
   conn.setRequestProperty("user-agent",
     "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
   // 发送POST请求必须设置如下两行
   conn.setDoOutput(true);
   conn.setDoInput(true);
   // 获取URLConnection对象对应的输出流
   out = new PrintWriter(conn.getOutputStream());
   // 发送请求参数
   out.print(param);
   // flush输出流的缓冲
   out.flush();
   // 定义BufferedReader输入流来读取URL的响应
   in = new BufferedReader(
     new InputStreamReader(conn.getInputStream()));
   String line;
   while ((line = in.readLine()) != null) {
    result += line;
   }
  } catch (Exception e) {
   System.out.println("发送 POST 请求出现异常!"+e);
   e.printStackTrace();
  }
  //使用finally块来关闭输出流、输入流
  finally{
   try{
    if(out!=null){
     out.close();
    }
    if(in!=null){
     in.close();
    }
   }
   catch(IOException ex){
    ex.printStackTrace();
   }
  }
  return result;
 } 
}

工具类有了之后,可以看到里面有两个发送请求的方法,一个是sendGet和sendPost方法,两个方法可以基于自己的选择选用,当然这只是个发送请求的方法,如何调用?在这里的话,我们既然是从es里面查询,就没有必要使用mybatis或者hibernate框架了,可以在数据层中,自己定义sql,然后将sql将拼好的sql通过前面的工具类调用sendGet/sendPost方法,如下:

String sql=URLEncoder.encode(" select * from table");encode方法主要是去除sql里面的一些空格

String result=HTTPSentUtils.sendGet("http://221.124.71.8:9200/_sql", "sql="+sql);ip加端口自己可以在安装的时候配置

OK!一般到这个时候的发送可以成功的话,就能取到es里面的数据,因为es里面返回的都是json数据,所以我们
格式化一下json字符串

net.sf.json.JSONObject jsonObject =net.sf.json.JSONObject.fromObject(result); 
//取出hits标签
net.sf.json.JSONObject hitsjsonObject = jsonObject.getJSONObject("hits");

数据拿到之后,结下来就是业务操作了。到这里也差不多结束了。

以上是查询的一些操作,那么如果我们需要将数据批量插入到es里面该如何操作?举个例子吧

package com.sojson.core.elasticsearch.manager;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import net.sf.json.JSONObject;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import com.sojson.common.utils.StringUtils;
import com.sojson.core.elasticsearch.utils.ESTools;
publicclassInsertManager{
/**
* 添加数据到Elasticsearch
*@param index 索引
*@param type 类型
*@param idName Id字段名称
*@param json 存储的JSON,可以接受Map
*@return
*/
publicstaticMapsave(String index,String type,String idName,JSONObject json){
List list =newArrayList();
list.add(json);
return save(index, type, idName, list);
}

通过传来的参数进行处理调用save方法执行插入es操作

/**
* 添加数据到Elasticsearch
*@param index 索引
*@param type 类型
*@param idName Id字段名称
*@param listData 一个对象集合
*@return
*/
@SuppressWarnings("unchecked")
publicstaticMapsave(String index,String type,String idName,List listData){
BulkRequestBuilder bulkRequest =ESTools.client.prepareBulk().setRefresh(true);
Map resultMap =newHashMap();
for(Object object: listData){
JSONObject json =JSONObject.fromObject(object);
//没有指定idName 那就让Elasticsearch自动生成,
if(StringUtils.isBlank(idName)){
IndexRequestBuilder lrb =ElasticsearchUtils.client
               .prepareIndex(index, type)
                .setSource(json);
bulkRequest.add(lrb);
//ElasticsearchUtils是工具类,里面配置的是一些es配置信息
}
else{
String idValue = json.optString(idName);
IndexRequestBuilder lrb =ESTools.client.prepareIndex(index, type,idValue).setSource(json);
bulkRequest.add(lrb);
}
}
BulkResponse bulkResponse = bulkRequest.execute().actionGet();
if(bulkResponse.hasFailures()){
// process failures by iterating through each bulk response item
System.out.println(bulkResponse.getItems().toString());
resultMap.put("500","保存ES失败!");
return resultMap;
}
bulkRequest=ESTools.client.prepareBulk();
resultMap.put("200","保存ES成功!");
return resultMap;
}
}

ElasticsearchUtils工具类

public class ElasticsearchUtils {
private static final String CLUSTER_NAME = "cluster.name";
private static final String ES_IP="es.ip";
private static final String ES_PORT="es.port";

private static Settings settings;
private static TransportClient client;
public static TransportClient getESClient() throws UnknownHostException{
settings = Settings.builder().put(CLUSTER_NAME,ConfigUtils.getConfig(CLUSTER_NAME)).build();
if(client != null){
client = new PreBuiltTransportClient(settings).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ConfigUtils.getConfig(ES_IP)), Integer.parseInt(ConfigUtils.getConfig(ES_PORT))));
}
return client;
}

以上是插入es操作,好了,今天的总结就到这里吧,希望可以给大家带来一些小小的帮助,也希望大家多多支持码农之家。

使用java操作elasticsearch的具体方法

系统环境: vm12 下的centos 7.2

当前安装版本: elasticsearch-2.4.0.tar.gz

Java操作es集群步骤1:配置集群对象信息;2:创建客户端;3:查看集群信息

1:集群名称

默认集群名为elasticsearch,如果集群名称和指定的不一致则在使用节点资源时会报错。

2:嗅探功能

通过client.transport.sniff启动嗅探功能,这样只需要指定集群中的某一个节点(不一定是主节点),然后会加载集群中的其他节点,这样只要程序不停即使此节点宕机仍然可以连接到其他节点。

3:查询类型SearchType.QUERY_THEN_FETCH

es 查询共有4种查询类型

QUERY_AND_FETCH:

主节点将查询请求分发到所有的分片中,各个分片按照自己的查询规则即词频文档频率进行打分排序,然后将结果返回给主节点,主节点对所有数据进行汇总排序然后再返回给客户端,此种方式只需要和es交互一次。

这种查询方式存在数据量和排序问题,主节点会汇总所有分片返回的数据这样数据量会比较大,二是各个分片上的规则可能不一致。

QUERY_THEN_FETCH:

主节点将请求分发给所有分片,各个分片打分排序后将数据的id和分值返回给主节点,主节点收到后进行汇总排序再根据排序后的id到对应的节点读取对应的数据再返回给客户端,此种方式需要和es交互两次。

这种方式解决了数据量问题但是排序问题依然存在而且是es的默认查询方式

DEF_QUERY_AND_FETCH: 和 DFS_QUERY_THEN_FETCH:

将各个分片的规则统一起来进行打分。解决了排序问题但是DFS_QUERY_AND_FETCH仍然存在数据量问题,DFS_QUERY_THEN_FETCH两种噢乖你问题都解决但是效率是最差的。

1, 获取client, 两种方式获取

@Before
 public void before() throws Exception {
  Map<String, String> map = new HashMap<String, String>(); 
  map.put("cluster.name", "elasticsearch_wenbronk"); 
  Settings.Builder settings = Settings.builder().put(map); 
  client = TransportClient.builder().settings(settings).build() 
      .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("www.wenbronk.com"), Integer.parseInt("9300"))); 
 }
@Before
 public void before11() throws Exception {
  // 创建客户端, 使用的默认集群名, "elasticSearch"
//  client = TransportClient.builder().build()
//    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("www.wenbronk.com"), 9300));

  // 通过setting对象指定集群配置信息, 配置的集群名
  Settings settings = Settings.settingsBuilder().put("cluster.name", "elasticsearch_wenbronk") // 设置集群名
//    .put("client.transport.sniff", true) // 开启嗅探 , 开启后会一直连接不上, 原因未知
//    .put("network.host", "192.168.50.37")
    .put("client.transport.ignore_cluster_name", true) // 忽略集群名字验证, 打开后集群名字不对也能连接上
//    .put("client.transport.nodes_sampler_interval", 5) //报错,
//    .put("client.transport.ping_timeout", 5) // 报错, ping等待时间,
    .build();
   client = TransportClient.builder().settings(settings).build()
     .addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress("192.168.50.37", 9300)));
   // 默认5s
   // 多久打开连接, 默认5s
   System.out.println("success connect");
 }

PS: 官网给的2种方式都不能用, 需要合起来才能用, 浪费老子一下午...

其他参数的意义:

使用java操作elasticsearch的具体方法

代码:

package com.wenbronk.javaes;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkProcessor.Listener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.script.Script;
import org.junit.Before;
import org.junit.Test;

import com.alibaba.fastjson.JSONObject;

/**
 * 使用java API操作elasticSearch
 * 
 * @author 231
 *
 */
public class JavaESTest {

 private TransportClient client;
 private IndexRequest source;
 
 /**
  * 获取连接, 第一种方式
  * @throws Exception
  */
// @Before
 public void before() throws Exception {
  Map<String, String> map = new HashMap<String, String>(); 
  map.put("cluster.name", "elasticsearch_wenbronk"); 
  Settings.Builder settings = Settings.builder().put(map); 
  client = TransportClient.builder().settings(settings).build() 
      .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("www.wenbronk.com"), Integer.parseInt("9300"))); 
 }

/**
  * 查看集群信息
  */
 @Test
 public void testInfo() {
  List<DiscoveryNode> nodes = client.connectedNodes();
  for (DiscoveryNode node : nodes) {
   System.out.println(node.getHostAddress());
  }
 }
 
 /**
  * 组织json串, 方式1,直接拼接
  */
 public String createJson1() {
  String json = "{" +
    "\"user\":\"kimchy\"," +
    "\"postDate\":\"2013-01-30\"," +
    "\"message\":\"trying out Elasticsearch\"" +
   "}";
  return json;
 }
 
 /**
  * 使用map创建json
  */
 public Map<String, Object> createJson2() {
  Map<String,Object> json = new HashMap<String, Object>();
  json.put("user", "kimchy");
  json.put("postDate", new Date());
  json.put("message", "trying out elasticsearch");
  return json;
 }
 
 /**
  * 使用fastjson创建
  */
 public JSONObject createJson3() {
  JSONObject json = new JSONObject();
  json.put("user", "kimchy");
  json.put("postDate", new Date());
  json.put("message", "trying out elasticsearch");
  return json;
 }
 
 /**
  * 使用es的帮助类
  */
 public XContentBuilder createJson4() throws Exception {
  // 创建json对象, 其中一个创建json的方式
  XContentBuilder source = XContentFactory.jsonBuilder()
   .startObject()
    .field("user", "kimchy")
    .field("postDate", new Date())
    .field("message", "trying to out ElasticSearch")
   .endObject();
  return source;
 }
 
 /**
  * 存入索引中
  * @throws Exception
  */
 @Test
 public void test1() throws Exception {
  XContentBuilder source = createJson4();
  // 存json入索引中
  IndexResponse response = client.prepareIndex("twitter", "tweet", "1").setSource(source).get();
//  // 结果获取
  String index = response.getIndex();
  String type = response.getType();
  String id = response.getId();
  long version = response.getVersion();
  boolean created = response.isCreated();
  System.out.println(index + " : " + type + ": " + id + ": " + version + ": " + created);
 }

 /**
  * get API 获取指定文档信息
  */
 @Test
 public void testGet() {
//  GetResponse response = client.prepareGet("twitter", "tweet", "1")
//        .get();
  GetResponse response = client.prepareGet("twitter", "tweet", "1")
    .setOperationThreaded(false) // 线程安全
    .get();
  System.out.println(response.getSourceAsString());
 }
 
 /**
  * 测试 delete api
  */
 @Test
 public void testDelete() {
  DeleteResponse response = client.prepareDelete("twitter", "tweet", "1")
    .get();
  String index = response.getIndex();
  String type = response.getType();
  String id = response.getId();
  long version = response.getVersion();
  System.out.println(index + " : " + type + ": " + id + ": " + version);
 }
 
 /**
  * 测试更新 update API
  * 使用 updateRequest 对象
  * @throws Exception 
  */
 @Test
 public void testUpdate() throws Exception {
  UpdateRequest updateRequest = new UpdateRequest();
  updateRequest.index("twitter");
  updateRequest.type("tweet");
  updateRequest.id("1");
  updateRequest.doc(XContentFactory.jsonBuilder()
    .startObject()
    // 对没有的字段添加, 对已有的字段替换
     .field("gender", "male")
     .field("message", "hello")
    .endObject());
  UpdateResponse response = client.update(updateRequest).get();
  
  // 打印
  String index = response.getIndex();
  String type = response.getType();
  String id = response.getId();
  long version = response.getVersion();
  System.out.println(index + " : " + type + ": " + id + ": " + version);
 }
 
 /**
  * 测试update api, 使用client
  * @throws Exception 
  */
 @Test
 public void testUpdate2() throws Exception {
  // 使用Script对象进行更新
//  UpdateResponse response = client.prepareUpdate("twitter", "tweet", "1")
//    .setScript(new Script("hits._source.gender = \"male\""))
//    .get();
  
  // 使用XContFactory.jsonBuilder() 进行更新
//  UpdateResponse response = client.prepareUpdate("twitter", "tweet", "1")
//    .setDoc(XContentFactory.jsonBuilder()
//      .startObject()
//       .field("gender", "malelelele")
//      .endObject()).get();
  
  // 使用updateRequest对象及script
//  UpdateRequest updateRequest = new UpdateRequest("twitter", "tweet", "1")
//    .script(new Script("ctx._source.gender=\"male\""));
//  UpdateResponse response = client.update(updateRequest).get();
  
  // 使用updateRequest对象及documents进行更新
  UpdateResponse response = client.update(new UpdateRequest("twitter", "tweet", "1")
    .doc(XContentFactory.jsonBuilder()
      .startObject()
       .field("gender", "male")
      .endObject()
     )).get();
  System.out.println(response.getIndex());
 }
 
 /**
  * 测试update
  * 使用updateRequest
  * @throws Exception 
  * @throws InterruptedException 
  */
 @Test
 public void testUpdate3() throws InterruptedException, Exception {
  UpdateRequest updateRequest = new UpdateRequest("twitter", "tweet", "1")
   .script(new Script("ctx._source.gender=\"male\""));
  UpdateResponse response = client.update(updateRequest).get();
 }
 
 /**
  * 测试upsert方法
  * @throws Exception 
  * 
  */
 @Test
 public void testUpsert() throws Exception {
  // 设置查询条件, 查找不到则添加生效
  IndexRequest indexRequest = new IndexRequest("twitter", "tweet", "2")
   .source(XContentFactory.jsonBuilder()
    .startObject()
     .field("name", "214")
     .field("gender", "gfrerq")
    .endObject());
  // 设置更新, 查找到更新下面的设置
  UpdateRequest upsert = new UpdateRequest("twitter", "tweet", "2")
   .doc(XContentFactory.jsonBuilder()
     .startObject()
      .field("user", "wenbronk")
     .endObject())
   .upsert(indexRequest);
  
  client.update(upsert).get();
 }
 
 /**
  * 测试multi get api
  * 从不同的index, type, 和id中获取
  */
 @Test
 public void testMultiGet() {
  MultiGetResponse multiGetResponse = client.prepareMultiGet()
  .add("twitter", "tweet", "1")
  .add("twitter", "tweet", "2", "3", "4")
  .add("anothoer", "type", "foo")
  .get();
  
  for (MultiGetItemResponse itemResponse : multiGetResponse) {
   GetResponse response = itemResponse.getResponse();
   if (response.isExists()) {
    String sourceAsString = response.getSourceAsString();
    System.out.println(sourceAsString);
   }
  }
 }
 
 /**
  * bulk 批量执行
  * 一次查询可以update 或 delete多个document
  */
 @Test
 public void testBulk() throws Exception {
  BulkRequestBuilder bulkRequest = client.prepareBulk();
  bulkRequest.add(client.prepareIndex("twitter", "tweet", "1")
    .setSource(XContentFactory.jsonBuilder()
      .startObject()
       .field("user", "kimchy")
       .field("postDate", new Date())
       .field("message", "trying out Elasticsearch")
      .endObject()));
  bulkRequest.add(client.prepareIndex("twitter", "tweet", "2")
    .setSource(XContentFactory.jsonBuilder()
      .startObject()
       .field("user", "kimchy")
       .field("postDate", new Date())
       .field("message", "another post")
      .endObject()));
  BulkResponse response = bulkRequest.get();
  System.out.println(response.getHeaders());
 }
 
 /**
  * 使用bulk processor
  * @throws Exception 
  */
 @Test
 public void testBulkProcessor() throws Exception {
  // 创建BulkPorcessor对象
  BulkProcessor bulkProcessor = BulkProcessor.builder(client, new Listener() {
   public void beforeBulk(long paramLong, BulkRequest paramBulkRequest) {
    // TODO Auto-generated method stub
   }
   
   // 执行出错时执行
   public void afterBulk(long paramLong, BulkRequest paramBulkRequest, Throwable paramThrowable) {
    // TODO Auto-generated method stub
   }
   
   public void afterBulk(long paramLong, BulkRequest paramBulkRequest, BulkResponse paramBulkResponse) {
    // TODO Auto-generated method stub
   }
  })
  // 1w次请求执行一次bulk
  .setBulkActions(10000)
  // 1gb的数据刷新一次bulk
  .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
  // 固定5s必须刷新一次
  .setFlushInterval(TimeValue.timeValueSeconds(5))
  // 并发请求数量, 0不并发, 1并发允许执行
  .setConcurrentRequests(1)
  // 设置退避, 100ms后执行, 最大请求3次
  .setBackoffPolicy(
    BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
  .build();
  
  // 添加单次请求
  bulkProcessor.add(new IndexRequest("twitter", "tweet", "1"));
  bulkProcessor.add(new DeleteRequest("twitter", "tweet", "2"));
  
  // 关闭
  bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
  // 或者
  bulkProcessor.close();
 }
}

tes2代码:

package com.wenbronk.javaes;
import java.net.InetSocketAddress;
import org.apache.lucene.queryparser.xml.FilterBuilderFactory;
import org.elasticsearch.action.search.MultiSearchResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.Settings.Builder;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.sort.SortParseElement;
import org.junit.Before;
import org.junit.Test;
/**
 * 使用java API操作elasticSearch
 * search API
 * @author 231
 *
 */
public class JavaESTest2 {

 private TransportClient client;

 /**
  * 获取client对象
  */
 @Before
 public void testBefore() {
  Builder builder = Settings.settingsBuilder();
  builder.put("cluster.name", "wenbronk_escluster");
//    .put("client.transport.ignore_cluster_name", true);
  Settings settings = builder.build();
  
  org.elasticsearch.client.transport.TransportClient.Builder transportBuild = TransportClient.builder();
  TransportClient client1 = transportBuild.settings(settings).build();
  client = client1.addTransportAddress((new InetSocketTransportAddress(new InetSocketAddress("192.168.50.37", 9300))));
  System.out.println("success connect to escluster");
  
 }
 
 /**
  * 测试查询
  */
 @Test
 public void testSearch() {
//  SearchRequestBuilder searchRequestBuilder = client.prepareSearch("twitter", "tweet", "1");
//  SearchResponse response = searchRequestBuilder.setTypes("type1", "type2")
//       .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
//       .setQuery(QueryBuilders.termQuery("user", "test"))
//       .setPostFilter(QueryBuilders.rangeQuery("age").from(0).to(1))
//       .setFrom(0).setSize(2).setExplain(true)
//       .execute().actionGet();
  SearchResponse response = client.prepareSearch()
    .execute().actionGet();
//  SearchHits hits = response.getHits();
//  for (SearchHit searchHit : hits) {
//   for(Iterator<SearchHitField> iterator = searchHit.iterator(); iterator.hasNext(); ) {
//    SearchHitField next = iterator.next();
//    System.out.println(next.getValues());
//   }
//  }
  System.out.println(response);
 }
 
 /**
  * 测试scroll api
  * 对大量数据的处理更有效
  */
 @Test
 public void testScrolls() {
  QueryBuilder queryBuilder = QueryBuilders.termQuery("twitter", "tweet");
  
  SearchResponse response = client.prepareSearch("twitter")
  .addSort(SortParseElement.DOC_FIELD_NAME, SortOrder.ASC)
  .setScroll(new TimeValue(60000))
  .setQuery(queryBuilder)
  .setSize(100).execute().actionGet();
  
  while(true) {
   for (SearchHit hit : response.getHits().getHits()) {
    System.out.println("i am coming");
   }
   SearchResponse response2 = client.prepareSearchScroll(response.getScrollId())
    .setScroll(new TimeValue(60000)).execute().actionGet();
   if (response2.getHits().getHits().length == 0) {
    System.out.println("oh no=====");
    break;
   }
  }
  
 }
 
 /**
  * 测试multiSearch
  */
 @Test
 public void testMultiSearch() {
  QueryBuilder qb1 = QueryBuilders.queryStringQuery("elasticsearch");
  SearchRequestBuilder requestBuilder1 = client.prepareSearch().setQuery(qb1).setSize(1);
  
  QueryBuilder qb2 = QueryBuilders.matchQuery("user", "kimchy");
  SearchRequestBuilder requestBuilder2 = client.prepareSearch().setQuery(qb2).setSize(1);
  
  MultiSearchResponse multiResponse = client.prepareMultiSearch().add(requestBuilder1).add(requestBuilder2)
    .execute().actionGet();
  long nbHits = 0;
  for (MultiSearchResponse.Item item : multiResponse.getResponses()) {
   SearchResponse response = item.getResponse();
   nbHits = response.getHits().getTotalHits();
   SearchHit[] hits = response.getHits().getHits();
   System.out.println(nbHits);
  }
  
 }
 
 /**
  * 测试聚合查询
  */
 @Test
 public void testAggregation() {
  SearchResponse response = client.prepareSearch()
    .setQuery(QueryBuilders.matchAllQuery()) // 先使用query过滤掉一部分
    .addAggregation(AggregationBuilders.terms("term").field("user"))
    .addAggregation(AggregationBuilders.dateHistogram("agg2").field("birth")
     .interval(DateHistogramInterval.YEAR))
    .execute().actionGet();
  Aggregation aggregation2 = response.getAggregations().get("term");
  Aggregation aggregation = response.getAggregations().get("agg2");
//  SearchResponse response2 = client.search(new SearchRequest().searchType(SearchType.QUERY_AND_FETCH)).actionGet();
 }
 
 /**
  * 测试terminate
  */
 @Test
 public void testTerminateAfter() {
  SearchResponse response = client.prepareSearch("twitter").setTerminateAfter(1000).get();
  if (response.isTerminatedEarly()) {
   System.out.println("ternimate");
  }
 }
 
 /**
  * 过滤查询: 大于gt, 小于lt, 小于等于lte, 大于等于gte
  */
 @Test
 public void testFilter() {
  SearchResponse response = client.prepareSearch("twitter") 
    .setTypes("") 
    .setQuery(QueryBuilders.matchAllQuery()) //查询所有 
    .setSearchType(SearchType.QUERY_THEN_FETCH) 
//    .setPostFilter(FilterBuilders.rangeFilter("age").from(0).to(19) 
//      .includeLower(true).includeUpper(true)) 
//    .setPostFilter(FilterBuilderFactory .rangeFilter("age").gte(18).lte(22)) 
    .setExplain(true) //explain为true表示根据数据相关度排序,和关键字匹配最高的排在前面 
    .get(); 
 }
 
 /**
  * 分组查询
  */
 @Test
 public void testGroupBy() {
  client.prepareSearch("twitter").setTypes("tweet")
  .setQuery(QueryBuilders.matchAllQuery())
  .setSearchType(SearchType.QUERY_THEN_FETCH)
  .addAggregation(AggregationBuilders.terms("user")
    .field("user").size(0)  // 根据user进行分组
           // size(0) 也是10
  ).get();
 } 
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持码农之家。

以上就是本次给大家分享的关于java的全部知识点内容总结,大家还可以在下方相关文章里找到相关文章进一步学习,感谢大家的阅读和支持。

elasticsearch 相关电子书
学习笔记
网友NO.498690

Spring Boot整合ElasticSearch实现多版本兼容的方法详解

前言 在上一篇学习SpringBoot中,整合了Mybatis、Druid和PageHelper并实现了多数据源的操作。本篇主要是介绍和使用目前最火的搜索引擎ElastiSearch,并和SpringBoot进行结合使用。 ElasticSearch介绍 ElasticSearch是一个基于Lucene的搜索服务器,其实就是对Lucene进行封装,提供了 REST API 的操作接口 ElasticSearch作为一个高度可拓展的开源全文搜索和分析引擎,可用于快速地对大数据进行存储,搜索和分析。 ElasticSearch主要特点:分布式、高可用、异步写入、多API、面向文档 。 ElasticSearch核心概念:近实时,集群,节点(保存数据),索引,分片(将索引分片),副本(分片可设置多个副本) 。它可以快速地储存、搜索和分析海量数据。 ElasticSearch使用案例:维基百科、Stack Overflow、Github 等等。 SpringBoot整合Elasticsearch 在使用SpringBoot整合Elasticsearch 之前,我们应该了解下它们……

网友NO.236117

使用ElasticSearch6.0快速实现全文搜索功能的示例代码

本文不涉及ElasticSearch具体原理,只记录如何快速的导入mysql中的数据进行全文检索。 工作中需要实现一个搜索功能,并且导入现有数据库数据,组长推荐用ElasticSearch实现,网上翻一通教程,都是比较古老的文章了,无奈只能自己摸索,参考ES的文档,总算是把服务搭起来了,记录下,希望有同样需求的朋友可以少走弯路,能按照这篇教程快速的搭建一个可用的ElasticSearch服务。 ES的搭建 ES搭建有直接下载zip文件,也有docker容器的方式,相对来说,docker更适合我们跑ES服务。可以方便的搭建集群或建立测试环境。这里使用的也是容器方式,首先我们需要一份Dockerfile: FROM docker.elastic.co/elasticsearch/elasticsearch-oss:6.0.0# 提交配置包括新的elasticsearch.yml 和keystore.jks文件COPY --chown=elasticsearch:elasticsearch conf/ /usr/share/elasticsearch/config/# 安装ikRUN ./bin/elasticsearch-plugin inst……

网友NO.236352

elasticsearch启动警告无法锁定JVM内存

elasticsearch启动警告 Unable to lock JVM memory (ENOMEM). This can result in part of the JVM being swapped out. Increase RLIMIT_MEMLOCK (ulimit). 内存锁定值的限制(max locked memory) 这个值只对普通用户起作用,对超级用户不起作用,这个问题是由于CAP_IPC_LOCK造成的.linux对内存是分页管理的,这意味着有不需要时,在物理内存的数据会被换到交换区或磁盘上.有需要时会被交换到物理内存,而将数据锁定到物理内存可以避免数据的换入/换出.采用锁定内存有两个理由:1)由于程序设计上需要,比如oracle等软件,就需要将数据锁定到物理内存.2)主要是安全上的需要,比如用户名和密码等等,被交换到swap或磁盘,有泄密的可能,所以一直将其锁定到物理内存. 查看项目进程限制,17497为pid cat /proc/17497/limits 查看系统限制ulimit -a core file size (blocks, -c) unlimiteddata seg size (kbytes, -d) unlimitedscheduling priority (-e) 0file size……

网友NO.808617

Spring Boot与Kotlin 整合全文搜索引擎Elasticsearch的示例代码

Elasticsearch 在全文搜索里面基本是无敌的,在大数据里面也很有建树,完全可以当nosql(本来也是nosql)使用。 这篇文章简单介绍Spring Boot使用Kotlin语言连接操作 Elasticsearch。但是不会做很详细的介绍,如果要深入了解Elasticsearch在Java/kotlin中的使用,请参考我之前编写的《Elasticsearch Java API 手册》 https://gitee.com/quanke/elasticsearch-java/ 里面包含使用实例,包含我们使用踩过的坑。 如果完全不了解Elasticsearch请先了解,安装好Elasticsearch服务 有多种方式连接Elasticsearch Spring Data Elasticsearch elasticsearch Java client 其他第三方库 需要注意的是,如果使用Spring Data Elasticsearch,spring boot 1.5++版本的不支持最新版本的elasticsearch。 下面是spring data elasticsearch 和elasticsearch对应的版本 spring data elasticsearch elasticsearch 3.0.0.RC2 5.5.0 3.0.0.M4 5.4.0 2.0.4.RELEASE 2.4.0 2.0.0.RELEASE 2.2.0 1.4.0.M1 1.7.3 ……

<
1
>

Copyright 2018-2020 www.xz577.com 码农之家

版权投诉 / 书籍推广 / 赞助:520161757@qq.com