java操作 elasticsearch进行增删改查-蒲公英云

本文中所使用的 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;
    }
}