文档章节

对ElasticSearch读写操作的封装

林中漫步
 林中漫步
发布于 2016/06/30 12:43
字数 1513
阅读 257
收藏 1
点赞 1
评论 0

1 依赖的jar包:

        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>1.7.1</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.4</version>
        </dependency>

2 Es环境配置:elasticsearch.yml

cluster.name: ${es.cluster.name}
node.name: "${es.node.name}"

client.transport.ip: ${es.client.transport.ip}
client.transport.port: ${es.client.transport.port}

3 封装的客户端工具类 ESClient

import com.alibaba.fastjson.JSONObject;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.FilterBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class ESClient {

    private static final Logger log = LoggerFactory.getLogger(ESClient.class);

    private TransportClient client;

    public ESClient() {
        this.init();
    }


    private void close() {
        if (this.client != null) {
            this.client.close();
        }
    }

    @Override
    public void finalize() throws Throwable {
        this.close();
        super.finalize();
    }

    private void init() {
        try {
            Settings settings = ImmutableSettings.settingsBuilder().loadFromClasspath("elasticsearch.yml")
                    .put("client.transport.sniff", true).build();
            this.client = new TransportClient(settings);

            int port = settings.getAsInt("client.transport.port", 9900);
            String[] ips = settings.getAsArray("client.transport.ip");

            for (String ip : ips) {
                log.info("the ip is:" + ip);
                client.addTransportAddress(new InetSocketTransportAddress(ip, port));
            }
            log.info("es连接成功:{},{}", client, JSONObject.toJSONString(client.listedNodes()));

        } catch (Exception e) {
            if (client != null) {
                client.close();
            }
            log.error("连接es失败!", e);
        }
    }

    /**
     * 为一份文档建立索引
     *
     * @param index 索引名,相当于关系型数据库的库名
     * @param type  文档类型,相当于关系型数据库的表名
     * @param json  json格式的数据集,必须含有属性"id"
     * @return
     */
    public IndexResponse indexDoc(String index, String type, String json) throws Exception {
        JSONObject kvMap = JSONObject.parseObject(json);
        return this.indexDoc(index, type, kvMap);
    }


    /**
     * 为一份文档建立索引
     *
     * @param index 索引名,相当于关系型数据库的库名
     * @param type  文档类型,相当于关系型数据库的表名
     * @param kvMap 键值对形式的数据集,map中必须有属性key: "id"
     * @return
     */
    public IndexResponse indexDoc(String index, String type, Map<String, Object> kvMap)
            throws Exception {
        if (!kvMap.containsKey("id")) {
            throw new Exception("创建索引时,传入的map或json串中没有属性'id'! ");
        }
        String id = (String) kvMap.get("id");
        if (id == null) {
            throw new Exception("创建索引时,传入的map或json的属性'id'的值为null! ");
        }

        IndexRequestBuilder builder = client.prepareIndex(index, type, id);
        IndexResponse response = builder.setSource(kvMap)
                .execute()
                .actionGet();
        return response;
    }


    /**
     * 为多份文档建立索引
     *
     * @param index 索引名,相当于关系型数据库的库名
     * @param type  文档类型,相当于关系型数据库的表名
     * @param jsons json格式的数据集,其下json串必须有属性"id"
     * @return
     */
    public BulkResponse batchIndexDocsForJson(String index, String type, List<String> jsons)
            throws Exception {
        if (jsons.isEmpty()) {
            throw new Exception("批量创建索引时,传入的参数'jsons'为空!");
        }

        List<Map<String, Object>> kvList = new ArrayList<Map<String, Object>>(jsons.size());
        for (String json : jsons) {
            JSONObject kvMap = JSONObject.parseObject(json);
            kvList.add(kvMap);
        }

        BulkResponse response = this.batchIndexDocsForMap(index, type, kvList);
        kvList.clear();
        return response;
    }


    /**
     * 为多份文档建立索引
     *
     * @param index  索引名,相当于关系型数据库的库名
     * @param type   文档类型,相当于关系型数据库的表名
     * @param kvList 键值对形式的数据集,其下map中必须有属性key: "id"
     * @return
     */
    public BulkResponse batchIndexDocsForMap(String index, String type, List<Map<String, Object>> kvList)
            throws Exception {
        if (kvList.isEmpty()) {
            throw new Exception("批量创建索引时,传入的参数'kvList'为空!");
        }

        List<IndexRequest> requestList = new ArrayList<IndexRequest>(kvList.size());

        for (Map<String, Object> kvMap : kvList) {
            if (!kvMap.containsKey("id")) {
                throw new Exception("批量创建索引时,传入的map或json串中没有属性'id'! ");
            }
            String id = (String) kvMap.get("id");
            if (id == null) {
                throw new Exception("批量创建索引时,传入的map或json的属性'id'的值为null! ");
            }

            IndexRequest request = client
                    .prepareIndex(index, type, id).setSource(kvMap)
                    .request();
            requestList.add(request);
        }

        BulkRequestBuilder bulkRequest = client.prepareBulk();
        for (IndexRequest request : requestList) {
            bulkRequest.add(request);
        }

        BulkResponse response = bulkRequest
                .execute()
                .actionGet();

        return response;
    }


    /**
     * 删除一个文档
     *
     * @param index 索引名,相当于关系型数据库的库名
     * @param type  文档类型,相当于关系型数据库的表名
     * @param id    键值对形式的数据集
     * @return
     */
    public DeleteResponse deleteDoc(String index, String type, String id) throws InterruptedException {
        DeleteRequestBuilder builder = client.prepareDelete(index, type, id);
        DeleteResponse response = builder
                .execute()
                .actionGet();
        return response;
    }

    /**
     * 根据条件删除多个文档
     *
     * @param index        索引名,相当于关系型数据库的库名
     * @param type         文档类型,相当于关系型数据库的表名
     * @param queryBuilder 查询器
     * @return
     */
    public void deleteDocsByQuery(String index, String type, QueryBuilder queryBuilder) {
        client.prepareDeleteByQuery(index).setTypes(type).setQuery(queryBuilder)
                .execute()
                .actionGet();
    }


    /**
     * 指定id获取文档
     *
     * @param index 索引名,相当于关系型数据库的库名
     * @param type  文档类型,相当于关系型数据库的表名
     * @param id    文档id
     * @return
     */
    public Map<String, Object> getDoc(String index, String type, String id) {
        GetResponse response = client.prepareGet(index, type, id)
                .execute()
                .actionGet();

        Map<String, Object> retMap = response.getSourceAsMap();
        return retMap;
    }


    public List<Map<String, Object>> search(String index, String type, QueryBuilder queryBuilder, FilterBuilder filterBuilder) {
        SearchRequestBuilder builder = client.prepareSearch(index).setTypes(type);
        if (queryBuilder != null) {
            builder = builder.setQuery(queryBuilder);
        }
        if (filterBuilder != null) {
            builder = builder.setPostFilter(filterBuilder);
        }

        SearchResponse searchResponse = builder.execute().actionGet();

        SearchHits hits = searchResponse.getHits();
        log.info("Es Hits count: " + hits.getTotalHits());

        List<Map<String, Object>> kvList = new ArrayList<Map<String, Object>>();

        SearchHit[] hitArray = hits.getHits();
        if (hitArray.length > 0) {
            for (SearchHit hit : hitArray) {
                Map<String, Object> kvMap = hit.getSource();
                kvMap.put("id", hit.getId());
                kvList.add(kvMap);
            }
        }
        return kvList;
    }
}

4 测试类 ESClientTest

import com.alibaba.fastjson.JSONObject;
import org.elasticsearch.index.query.FilterBuilder;
import org.elasticsearch.index.query.FilterBuilders;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class ESClientTest {

    private static final Logger log = LoggerFactory.getLogger(ESClientTest.class);

    @Test
    public void testIndexDoc() {
        ESClient client = new ESClient();
        try {
            // 创建索引
            Map<String, Object> kvMap = new HashMap<String, Object>();
            kvMap.put("name", "hu");
            kvMap.put("age", "30");
            kvMap.put("gender", "f");
            kvMap.put("id", "33sdfa");
            client.indexDoc("test", "test_user", kvMap);

            // 创建索引
            String json = "{\"id\":\"55\",\"bb\":\"bbs\"}";
            client.indexDoc("test", "test_user", json);


            Thread.sleep(3000);

            // 查询结果
            QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();
            List<Map<String, Object>> kvList = client.search("test", "test_user", queryBuilder, null);
            for (Map<String, Object> kv : kvList) {
                System.out.println(JSONObject.toJSONString(kv));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void testBatchIndexDocsForJson() {
        String json1 = "{\"id\":\"1\",\"ak\":\"av\",\"bk\":\"bv\"}";
        String json2 = "{\"id\":\"2\",\"ak\":\"av2\",\"bk\":\"bv2\"}";
        String json3 = "{\"id\":\"3\",\"ak\":\"av3\",\"bk\":\"bv3\"}";

        List<String> jsonList = new ArrayList<String>();
        jsonList.add(json1);
        jsonList.add(json2);
        jsonList.add(json3);

        ESClient client = new ESClient();
        try {
            // 创建索引
            client.batchIndexDocsForJson("test", "test_hw", jsonList);

            Thread.sleep(3000);

            // 查询结果
            QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();
            List<Map<String, Object>> kvList = client.search("test", "test_hw", queryBuilder, null);
            for (Map<String, Object> kv : kvList) {
                System.out.println(JSONObject.toJSONString(kv));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void testBatchIndexDocsForMap() {
        Map<String, Object> kvMap = new HashMap<String, Object>();
        kvMap.put("id", "3");
        kvMap.put("name", "hu");
        kvMap.put("age", "30");
        kvMap.put("gender", "f");
        kvMap.put("asdf", "33");

        Map<String, Object> kvMap2 = new HashMap<String, Object>();
        kvMap2.put("id", "2");
        kvMap2.put("name", "wang");
        kvMap2.put("age", "35");
        kvMap2.put("gender", "f");

        List<Map<String, Object>> kvList = new ArrayList<Map<String, Object>>();
        kvList.add(kvMap);
        kvList.add(kvMap2);


        ESClient client = new ESClient();
        try {
            // 创建索引
            client.batchIndexDocsForMap("test", "test_user", kvList);

            Thread.sleep(3000);

            // 查询结果
            QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();
            kvList = client.search("test", "test_user", queryBuilder, null);
            for (Map<String, Object> kv : kvList) {
                System.out.println(JSONObject.toJSONString(kv));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void testSearch() {
        ESClient client = new ESClient();
        try {

            QueryBuilder queryBuilder = QueryBuilders.termQuery("gender", "f");
            FilterBuilder filterBuilder = FilterBuilders.rangeFilter("age").from(32);

            List<Map<String, Object>> kvList = client.search("test", "test_user", queryBuilder, filterBuilder);
            for (Map<String, Object> kv : kvList) {
                System.out.println(JSONObject.toJSONString(kv));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void testDeleteDoc() {
        ESClient client = new ESClient();
        try {

            client.deleteDoc("test", "test_user", "1");

            Thread.sleep(3000);

            // 查询结果
            QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();
            List<Map<String, Object>> kvList = client.search("test", "test_hw", queryBuilder, null);
            for (Map<String, Object> kv : kvList) {
                System.out.println(JSONObject.toJSONString(kv));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void testDeleteDocs() {
        ESClient client = new ESClient();
        try {
            // 删除文档
            QueryBuilder queryBuilder = QueryBuilders.termQuery("ak", "av2");
            client.deleteDocsByQuery("test", "test_user", queryBuilder);

            // 查询结果
            queryBuilder = QueryBuilders.matchAllQuery();
            List<Map<String, Object>> kvList = client.search("test", "test_user", queryBuilder, null);
            for (Map<String, Object> kv : kvList) {
                System.out.println(JSONObject.toJSONString(kv));
            }

            queryBuilder = QueryBuilders.matchAllQuery();
            client.deleteDocsByQuery("test", "test_user", queryBuilder);

            queryBuilder = QueryBuilders.matchAllQuery();
            client.deleteDocsByQuery("test", "test_hw", queryBuilder);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

© 著作权归作者所有

共有 人打赏支持
林中漫步
粉丝 90
博文 39
码字总数 33158
作品 0
深圳
架构师
当ES赶超Redis,这份ES进修攻略不容错过!

从4月DB-Engines最新发布的全球数据库排名中,我们赫然发现ElasticSearch逆袭超越了Redis,从原先的第9名上升至第8名,而Redis则落后一名,排在了其后。 事实上,这场逆袭并不算太让人意外。...

DBAplus社群 ⋅ 04/15 ⋅ 0

快速上手 Elasticsearch 的几个建议

相信不少同学都听说过 Elasticsearch,作为目前最流行的搜索引擎实现方案,越来越多的公司在自己的架构中引入,而其应用场景也从搜索引擎扩展到了日志存储分析、大数据分析领域,本文尝试给初...

rockybean ⋅ 05/21 ⋅ 0

Elasticsearch 5.5.1的安装和入门教程(转)

说明:转自老阮的文章,业界最简单的入门教程。一切的安装的运行建议不要用root权限,最好是当前用户下的权限。 作者: 阮一峰 日期: 2017年8月17日 全文搜索属于最常见的需求,开源的 Elas...

easonjim ⋅ 01/21 ⋅ 0

Elasticsearch中文分词研究

一、ES分析器简介 ES是一个实时搜索与数据分析引擎,为了完成搜索功能,必须对原始数据进行分析、拆解,以建立索引,从而实现搜索功能; ES对数据分析、拆解过程如下: 首先,将一块文本分成...

zhaipengfei1231 ⋅ 04/18 ⋅ 0

Centos6搭建elk系统,监控IIS日志

**所需程序: 服务器端:java、elasticsearch、kikbana 客 户 端:IIS、logstash** 一、服务器端(192.168.10.46)操作: 先建立一个ELK专门的目录: [root@Cent65 ~]mkdir /elk/ 上传到elk...

D杀手D ⋅ 04/24 ⋅ 0

elasticsearch-head 安装介绍

elasticsearch-head 是用于监控 Elasticsearch 状态的客户端插件,包括数据可视化、执行增删改查操作等。elasticsearch-head 插件的安装在 Linux 和 Windows 没什么区别,安装之前确保当前系...

BeckJin ⋅ 05/19 ⋅ 0

spring-data-elasticsearch 基本案例详解(三)

『 风云说:能分享自己职位的知识的领导是个好领导。 』 运行环境:JDK 7 或 8,Maven 3.0+ 技术栈:SpringBoot 1.5+, Spring Data Elasticsearch 1.5+ ,ElasticSearch 2.3.2 本文提纲 一、...

夜黑人模糊灬 ⋅ 05/13 ⋅ 0

基于ELK实时日志分析的最佳实践

在2018云栖大会深圳峰会大数据分析与可视化专场上,由阿里巴巴搜索引擎事业部开放搜索团队的吴迪带来了“基于ELK实时日志分析的最佳实践”的主题分享。介绍了传统的日志分析、ELK的概念和ELK...

smile小太阳 ⋅ 05/06 ⋅ 0

分布式定时任务Elastic-Job框架在SpringBoot工程中的应用实践(二)

文章摘要:在生产环境中部署Elastic-Job集群后,那么如何来运维监控线上跑着的定时任务呢? 如果在生产环境的大规模服务器集群上部署了集成Elastic-Job的业务工程,而没有相应的运维监控工具...

癫狂侠 ⋅ 05/15 ⋅ 0

CTO详细讲解海量日志处理ELK

ELK实时日志分析平台之Elasticsearch简介 Elasticsearch是一个高度灵活的开源全文检索和分析引擎。它能够迅速(几乎是实时地)地存储、查找和分析大规模数据。通常被用在有复杂的搜索要求的系...

Java架构分享 ⋅ 05/23 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

JPA入门,配置文件的设置

<?xml version="1.0" encoding="UTF-8"?> <persistence xmlns="http://java.sun.com/xml/ns/persistence" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http......

码农屌丝 ⋅ 15分钟前 ⋅ 0

Java基础——面向对象和构造器

声明:本栏目所使用的素材都是凯哥学堂VIP学员所写,学员有权匿名,对文章有最终解释权;凯哥学堂旨在促进VIP学员互相学习的基础上公开笔记。 静态成员介绍 为什么要有静态成员?静态成员用来...

凯哥学堂 ⋅ 17分钟前 ⋅ 0

vmware中Centos 7 linux的LVM磁盘扩容

系统是RHEL7(centos7差不多一样) 关闭系统,在vmware、设置、硬盘、扩展、输入数字大于当前系统内存、点击扩展。 开机再查看磁盘信息 fdisk -l 注意:可以看出sda磁盘增加了,但是根目录还...

gugudu ⋅ 27分钟前 ⋅ 0

JAVA线程sleep和wait方法区别

昨天面试,突然被问到sleep 和 wait的区别,一下子有点蒙,在这里记一下,以示警戒。 首先说sleep,sleep就是正在执行的线程主动让出cpu,cpu去执行其他线程,在sleep指定的时间过去后,cpu...

徐玉强 ⋅ 29分钟前 ⋅ 0

vuex学习--模块

随着项目复杂性增加,共享状态也越来越多。需要对转态操作进行分组,分组后在进行分组编写。学习一下module:状态管理器的模块组操作。 首先是声明: const moduleA={ state,mutations,g...

大美琴 ⋅ 31分钟前 ⋅ 0

Selenium 简单入门

安装 pip install selenium 驱动下载 https://chromedriver.storage.googleapis.com/index.html 下载最新的驱动,放入path中,可以放入Python的scripts目录下,也可以放入Chrome安装目录,并...

阿豪boy ⋅ 33分钟前 ⋅ 0

292. Nim Game - LeetCode

Question 292. Nim Game Solution 思路:试着列举一下,就能发现一个n只要不是4的倍数,就能赢。 n 是否能赢1 true2 true3 true4 false 不论删除几,对方都能一把赢5 t...

yysue ⋅ 今天 ⋅ 0

6.5 zip压缩工具 6.6 tar打包 6.7 打包并压缩

zip压缩工具 zip命令可以压缩目录和文件,-r 压缩目录。 zip使用方法 zip 1.txt.zip 1.txt //压缩文件 zip -r 123.zip 123/ //压缩目录 unzip 1.txt.zip //解压 unzip 123.zip -d /root/456...

Linux_老吴 ⋅ 今天 ⋅ 0

react-loadable使用跳坑

官方给react-loadable的定义是: A higher order component for loading components with dynamic imports. 动态路由示例 withLoadable.js import React from 'react'import Loadable fro......

pengqinmm ⋅ 今天 ⋅ 0

记录工作中遇到的坑

1、ios safari浏览器向下滚动会触发window resize事件

端木遗风 ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部