ElasticSearch Java API的使用

Posted 2022-12-08 準提童子

先上代码，后续分析。

jar包直接copy的es包下的，实现的功能有：

1.创建索引

2.查询具体文档

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

public class ESAPITest 
	private Client client;
	
	public ESAPITest()
		//首先配置es集群名称，到Settings对象
		Settings settings = ImmutableSettings.settingsBuilder().put("cluster.name", "inde-test").build();
		//其次，基于Settings对象创建Clinet对象，连接9300
		this.client = new TransportClient(settings).
			addTransportAddress(new InetSocketTransportAddress("101.200.xxx.xxx", 9300));
	
	public static void main(String[] args)
		ESAPITest esApITest = new ESAPITest();
		esApITest.createIndex();
		esApITest.getIndex();
		
	
	//创建 索引+类型+文档
	public void  createIndex() 
		//生成索引,使用Map存储
		Map<String,Object> json = new HashMap<>();
		json.put("user", "cas");
		json.put("postDate", new Date());
		json.put("message", "try Es Java API");
		
		IndexResponse response = this.client.prepareIndex("es-api", "test", "2").setSource(json).execute().actionGet();
		System.out.println(response.isCreated());
		//创建成功为true;失败为false;如果里边已经有了该文档，在创建一次，返回false
		client.close();
	
	//根据 索引+类型+文档 获取信息
	public void getIndex()
		GetResponse getResponse = client.prepareGet("es-api","test","1")
				.execute().actionGet();
		Map<String,Object> res = getResponse.getSource();
		//查询不到，会返回空指针
		if(res == null)
			System.out.println("empty");
		else
			System.out.println(res.toString());
			//message=try Es Java API, postDate=2016-08-30T01:56:32.131Z, user=cas
		
		client.close();
	

BulkProcessor设置批量请求的属性

		//BulkProcessor
		BulkProcessor bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() 
			@Override
			public void beforeBulk(long arg0, BulkRequest arg1) 
				//批量执行前做的事情
				System.out.println("bulk api action starting...");
			
			@Override
			public void afterBulk(long arg0, BulkRequest arg1, Throwable arg2) 
				System.out.println("exception:bukl api action ending...:"+arg2.getMessage());
			
			@Override
			public void afterBulk(long arg0, BulkRequest arg1, BulkResponse arg2) 
				//正常执行完毕后...
				System.out.println("normal:bukl api action ending...");
			
		)
		//设置多种条件，对批量操作进行限制，达到限制中的任何一种触发请求的批量提交
		.setBulkActions(1000)//设置批量操作一次性执行的action个数，根据请求个数批量提交
		//.setBulkSize(new ByteSizeValue(1,ByteSizeUnit.KB))//设置批量提交请求的大小允许的最大值
		//.setFlushInterval(TimeValue.timeValueMillis(100))//根据时间周期批量提交请求
		//.setConcurrentRequests(1)//设置允许并发请求的数量
		//设置请求失败时的补偿措施，重复请求3次
		//.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
		.build();
		for(int i =0;i<100000;i++)
			bulkProcessor.add(new IndexRequest("index-test","weibo2",""+i).source(
					XContentFactory
					.jsonBuilder()
					.startObject()
					.field("name","yuchen"+i)
					.field("interest","love"+i)
					.endObject()));
		
		System.out.println("load succeed!");