java操作 elasticsearch进行增删改查

一时失言乱红尘 2022-05-26 11:20 315阅读 0赞

本文中所使用的 elasticsearch 版本是 5.1.1

elasticsearch版本不同，对其进行操作的API也不同。

具体可以详见elsticsearch官方文档，可以选择不同版本的javaapi，操作版本不同，其api也会有相应的差异

api见下面的网址

http://www.elastic.co/guide/en/elasticsearch/client/java-api/5.1/index.html

elasticsearch的启动和具体增删改语句见下面两篇博客：

安装：   https://blog.csdn.net/baidu\_24545901/article/details/79019812

增删改查：  https://blog.csdn.net/baidu\_24545901/article/details/79031291

接下来代码主要是 elasticsearch5.1.1版本的javaapi

package com.csii.webservice.dao;
    
    import java.io.IOException;
    import java.net.InetAddress;
    import java.net.InetSocketAddress;
    import java.net.UnknownHostException;
    import java.util.Date;
    import java.util.concurrent.ExecutionException;
    
    import org.elasticsearch.action.ActionListener;
    import org.elasticsearch.action.bulk.BulkRequestBuilder;
    import org.elasticsearch.action.bulk.BulkResponse;
    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.client.transport.TransportClient;
    import org.elasticsearch.common.settings.Settings;
    import org.elasticsearch.common.transport.InetSocketTransportAddress;
    import org.elasticsearch.common.xcontent.XContentBuilder;
    import org.elasticsearch.common.xcontent.XContentFactory;
    import org.elasticsearch.index.query.QueryBuilders;
    import org.elasticsearch.index.reindex.BulkIndexByScrollResponse;
    import org.elasticsearch.index.reindex.DeleteByQueryAction;
    import org.elasticsearch.script.Script;
    import org.elasticsearch.script.ScriptService;
    import org.elasticsearch.transport.client.PreBuiltTransportClient;
    import org.junit.Test;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Repository;
    
    import com.csii.webservice.entiry.Message;
    
    @Repository
    public class MessageDao {
    
    	public final static String host = "127.0.0.1";
    	public final static int port = 9300;
    	
    	private Logger logger = LoggerFactory.getLogger(MessageDao.class);
    	
    	
    	/*
    	 这个是 elasticsearch的增加索引并且增加数据
    	 */
    	public static void main1(String[] args) /*addIndex1()*/  {
    		try{
    			TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    			        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    			System.out.println("连接成功！！");
    			
    			IndexResponse response = client.prepareIndex("twitter", "tweet", "1")
    			        .setSource(XContentFactory.jsonBuilder()
    			                    .startObject()
    			                        .field("user", "kimchy")
    			                        .field("postDate", new Date())
    			                        .field("message", "trying out Elasticsearch")
    			                    .endObject()
    			                  )
    			        .get();
    			
    			System.out.println(
    					response.getId()+"  "+
    					response.getType()+"   "+
    					response.getVersion()
    					);
    			
    			
    		}catch(Exception e){
    			e.printStackTrace();
    		}
    	}
    	
    	/*
    	 这个是增加索引的第二种形式
    	 */
    	public static void main2(String[] args) {
    		try{
    			TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    			        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    			System.out.println("连接成功！！");
    			
    			String json = "{" +
    			        "\"user\":\"kimchy\"," +
    			        "\"postDate\":\"2013-01-30\"," +
    			        "\"message\":\"trying out Elasticsearch\"" +
    			    "}";
    
    			IndexResponse response = client.prepareIndex("twitter1", "tweet1")
    			        .setSource(json)
    			        .get();
    			
    			System.out.println(
    					response.getId()+"  "+
    					response.getType()+"   "+
    					response.getVersion()
    					);
    			
    			
    		}catch(Exception e){
    			e.printStackTrace();
    		}
    	}
    	
    	/*
    	获取es中数据的api
    	*/
    	public static void main3(String[] args) {
    		try{
    			TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    			        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    			System.out.println("连接成功！！");
    			
    			GetResponse response = client.prepareGet("twitter", "tweet", "1").get();
    			
    			System.out.println(response);
    			
    			GetResponse response1 = client.prepareGet("twitter", "tweet", "1")
    			        .setOperationThreaded(false)
    			        .get();
    			System.out.println(response1);
    			
    		}catch(Exception e){
    			e.printStackTrace();
    		}
    	}
    	
    	/*
    	 测试es删除的 api， 删除的是json数据，索引不删除
    	 */
    	public static void main4(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		System.out.println("连接成功！！");
    		
    		DeleteResponse response = client.prepareDelete("twitter", "tweet", "1").get();
    		System.out.println(response);
    	}
    	
    	/*
    	删除elasticsearch当中的index中的数据
    	 */
    	public static void main5(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		System.out.println("连接成功！！");
    		
    		BulkIndexByScrollResponse response =
    				DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
    				 .filter(QueryBuilders.matchQuery("title", "爬"))    //先通过一个查询规则查询
    				 .source("twitter")                                  //查询的 index	
    				 .get();                                             //获取到查询的返回	
    	long deleted = response.getDeleted();  				//获取到 一共删除的总条数
    	
    	System.out.println(deleted);
    		
    	}
    	
    	/*
    	 删除es中的数据
    	 同时在删除的时候加上了事件的监听
    	 */
    	public static void main6(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
    	    .filter(QueryBuilders.matchQuery("title", "java"))                  
    	    .source("twitter")                                                   
    	    .execute(new ActionListener<BulkIndexByScrollResponse>() {           
    	      
    	        public void onResponse(BulkIndexByScrollResponse response) {
    	            long deleted = response.getDeleted();                        
    	            System.out.println(deleted);	
    	        }
    	        
    	        public void onFailure(Exception e) {
    	            // Handle the exception
    	        	System.out.println("删除失败");
    	        }
    	    });
    	}
    	
    	/*
    	 执行es修改数据的操作
    	 */
    	public static void main7(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		UpdateRequest updateRequest = new UpdateRequest();
    		updateRequest.index("twitter");
    		updateRequest.type("tweet");
    		updateRequest.id("1");
    		updateRequest.doc(XContentFactory.jsonBuilder()
    		        .startObject()
    		            .field("salary_min", "16000")
    		        .endObject());
    		client.update(updateRequest).get();
    		
    	}
    	
    	/*
    	 执行es修改数据的操作
    	 使用  client 的 prepareUpdate这个方法
    	 */
    	public static void main8(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		client.prepareUpdate("twitter", "tweet", "1")
            .setDoc(XContentFactory.jsonBuilder()               
                .startObject()
                    .field("salary_min", "17000")
                .endObject())
            .get();
    		
    	}
    	
    	/*
    	 通过传json字符串的形式进行更新操作
    	 */
    	public static void main9(String[] args) throws Exception, ExecutionException {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		UpdateRequest updateRequest = new UpdateRequest("twitter", "tweet", "1")
    		        .script(new Script("ctx._source.salary_min = \"19000\""));
    		client.update(updateRequest).get();
    	}
    	
    	/*
    	 插入或者更新的操作
    	 当id值记录存在的时候执行修改操作，
    	 当id值不存在的时候执行插入操作
    	 */
    	public static void main10(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		IndexRequest indexRequest = new IndexRequest("twitter", "tweet", "4")
    		        .source(XContentFactory.jsonBuilder()
    		            .startObject()
    		                .field("salary_min", "20000")
    		            .endObject());
    		
    		UpdateRequest updateRequest = new UpdateRequest("twitter", "tweet", "4")
    		        .doc(XContentFactory.jsonBuilder()
    		            .startObject()
    		                .field("salary_min", "10000")
    		            .endObject())
    		        .upsert(indexRequest);              
    		client.update(updateRequest).get();
    	}
    	
    	
    	/*
    	 多元化的get
    	 */
    	public static void main(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		MultiGetResponse multiGetItemResponses = client.prepareMultiGet()
    			    .add("twitter", "tweet", "3")     //通过单独id get      
    			    .add("twitter", "tweet", "3", "4") //通过 list 的id get
    			    .add("twitter1", "tweet", "3")          //也可以通过另外一个索引
    			    .get();
    
    			for (MultiGetItemResponse itemResponse : multiGetItemResponses) { 
    			    GetResponse response = itemResponse.getResponse();
    			    if (response.isExists()) {                      
    			    	String json = response.getSourceAsString(); 
    			    	System.out.println("-------------------");
    			    	System.out.println( json );
    			    }
    			}
    		
    	}
    	
    	/*
    	 通过bulk来操作 es,添加数据
    	 */
    	public static void main12(String[] args) throws Exception {
    		TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    		        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
    		
    		BulkRequestBuilder bulkRequest = client.prepareBulk();
    		
    		bulkRequest.add(client.prepareIndex("twitter", "tweet", "3")
    		        .setSource(XContentFactory.jsonBuilder()
    		                    .startObject()
    		                        .field("user", "kimchy")
    		                        .field("postDate", new Date())
    		                        .field("message", "trying out Elasticsearch")
    		                    .endObject()
    		                  )
    		        );
    
    		bulkRequest.add(client.prepareIndex("twitter", "tweet", "5")
    		        .setSource(XContentFactory.jsonBuilder()
    		                    .startObject()
    		                        .field("user", "kimchy")
    		                        .field("postDate", new Date())
    		                        .field("message", "another post")
    		                    .endObject()
    		                  )
    		        );		
    		
    	
    		
    	}
    	
    	
    	public int addMessage(Message message) {
    		// TODO Auto-generated method stub
    		
    		try {
    			TransportClient client = new PreBuiltTransportClient(Settings.EMPTY).addTransportAddresses(
    		                               new InetSocketTransportAddress(InetAddress.getByName(host),port));
    			BulkIndexByScrollResponse response =
    				    DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
    				        .filter(QueryBuilders.matchQuery("title", "爬")) 
    				        .source("twitter")                                  
    				        .get();                                             
    
    				long deleted = response.getDeleted();  
    			
    			System.out.println(deleted);
    				
    		} catch (UnknownHostException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		
    		return 1;
    		
    	}
    
    	
    	
    }