`

APDPlat拓展搜索之集成ElasticSearch

APDPlatElasticSearch开发平台JAVA 
阅读更多

APDPlat充分利用Compass的OSEM和ORM integration特性，提供了简单易用且功能强大的内置搜索特性。

 

APDPlat的内置搜索，在设计简洁优雅的同时，还具备了强大的实时搜索能力，用户只需用注解的方式在模型中指定需要搜索哪些字段（还可在模型之间进行关联搜索）就获得了搜索能力，而不用编写任何代码。平台自动处理索引维护、查询解析、结果高亮等支撑功能。

 

然而APDPlat的内置搜索只能在单机上面使用，不支持分布式，只能用于中小规模的场景。为了支持大规模的分布式搜索和实时分析，APDPlat选用Compass的进化版ElasticSearch （Compass和ElasticSearch的关系）。

 

ElasticSearch提供了Java Client API，但是由于该API依赖于Lucene的org.apache.lucene.util包中的几个类，以致于无法和APDPlat集成，原因是APDPlat中Compass依赖的Lucene的版本和ElasticSearch依赖的版本冲突。

 

从这里可以得知，ElasticSearch的Java Client API如果完全移除对Lucene的依赖，仅仅作为用户和ElasticSearch集群之间通信的接口，使用起来就会更方便。

 

因此，APDPlat只能采用ElasticSearch的RESTful API。

 

接下来我们看一个APDPlat和ElasticSearch集成的例子：

 

APDPlat提供了可扩展的日志处理接口，用户可编写自己的插件并在配置文件中指定启用哪些插件，日志处理接口如下：

 

/**
 * 日志处理接口:
 * 可将日志存入独立日志数据库（非业务数据库）
 * 可将日志传递到activemq\rabbitmq\zeromq等消息队列
 * 可将日志传递到kafka\flume\chukwa\scribe等日志聚合系统
 * 可将日志传递到elasticsearch\solr等搜索服务器
 * @author 杨尚川
 */
public interface LogHandler {
    public <T extends Model> void handle(List<T> list);
}

 

将日志传递到ElasticSearch搜索服务器的实现使用了几个配置信息，这些配置信息默认存放在config.properties中，如下所示：

 

#elasticsearch服务器配置
elasticsearch.host=localhost
elasticsearch.port=9200
elasticsearch.log.index.name=apdplat_for_log

 

因为LogHandler接口中定义的参数List<T> list为泛型，只知道T是Model的子类，而不知道具体是哪一个类，所以我们使用反射的机制来获取具体对象类型：

 

String simpleName = model.getClass().getSimpleName();
LOG.debug((j++)+"、simpleName: 【"+simpleName+"】");
json.append("{\"index\":{\"_index\":\"")
  .append(INDEX_NAME)
  .append("\",\"_type\":\"")
  .append(simpleName)
  .append("\"}}")
  .append("\n");
json.append("{");

 

同时，我们利用反射的方式获取对象的字段以及相应的值，并正确处理类型问题：

 

Field[] fields = model.getClass().getDeclaredFields();
int len = fields.length;
for(int i = 0; i < len; i++){
	Field field = fields[i];
	String name = field.getName();
	field.setAccessible(true);
	Object value = field.get(model);
	//小心空指针异常，LogHandler线程会悄无声息地退出！
	if(value == null){
		LOG.debug("忽略空字段："+name);
		continue;
	}
	if(i>0){
		json.append(",");
	}
	String valueClass=value.getClass().getSimpleName();
	LOG.debug("name: "+name+"   type: "+valueClass);
	if("Timestamp".equals(valueClass) || "Date".equals(valueClass)){
		//提交给ES的日期时间值要为"2014-01-31T13:53:54"这样的形式
		value=DateTypeConverter.toDefaultDateTime((Date)value).replace(" ", "T");
	}
	String prefix = "\"";
	String suffix = "\"";
	//提交给ES的数字和布尔值不要加双引号
	if("Float".equals(valueClass)
			|| "Double".equals(valueClass) 
			|| "Long".equals(valueClass) 
			|| "Integer".equals(valueClass)
			|| "Short".equals(valueClass)
			|| "Boolean".equals(valueClass)){
		prefix="";
		suffix="";
	}
	json.append("\"")
			.append(name)
			.append("\":")
			.append(prefix)
			.append(value)
			.append(suffix);
}
json.append("}\n");

 

构造完要提交的JSON数据之后，向服务器发送HTTP PUT请求：

 

HttpURLConnection conn = (HttpURLConnection) URL.openConnection();
conn.setRequestMethod("PUT");
conn.setDoOutput(true);
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(conn.getOutputStream(),"utf-8"));    
writer.write(json.toString());
writer.flush();
StringBuilder result = new StringBuilder();
try (BufferedReader reader = new BufferedReader (new InputStreamReader (conn.getInputStream()))) {
	String line = reader.readLine();
	while(line != null){
		result.append(line);
		line = reader.readLine();
	}
}

 

服务器会以JSON数据格式返回处理结果，我们使用Jackson解析返回的JSON字符串：

 

JsonNode node = MAPPER.readTree(resultStr);
for(JsonNode item : node.get("items")){
	JsonNode createJsonNode = item.get("create");
	JsonNode okJsonNode = createJsonNode.get("ok");
	if(okJsonNode != null){
		boolean r = okJsonNode.getBooleanValue();
		if(r){
			success++;
		}
	}else{
		JsonNode errorJsonNode = createJsonNode.get("error");
		if(errorJsonNode != null){
			String errorMessage = errorJsonNode.getTextValue();
			LOG.error("索引失败："+errorMessage);
		}
	}
}

 

下面是ElasticSearchLogHandler完整的实现：

 

/**
 * 
 * 日志处理实现:
 * 将日志保存到ElasticSearch中
 * 进行高性能实时搜索和分析
 * 支持大规模分布式搜索
 * 
 * @author 杨尚川
 */
@Service
public class ElasticSearchLogHandler implements LogHandler{
    private static final APDPlatLogger LOG = new APDPlatLogger(ElasticSearchLogHandler.class);
    
    private static final String INDEX_NAME = PropertyHolder.getProperty("elasticsearch.log.index.name");
    private static final String HOST = PropertyHolder.getProperty("elasticsearch.host");
    private static final String PORT = PropertyHolder.getProperty("elasticsearch.port");
    private static final ObjectMapper MAPPER = new ObjectMapper();
    private static URL URL;
    
    private int success;
   
    public ElasticSearchLogHandler(){
        LOG.info("elasticsearch.log.index.name: "+INDEX_NAME);
        LOG.info("elasticsearch.host: "+HOST);
        LOG.info("elasticsearch.port: "+PORT);
        try {
            URL = new URL("http://"+HOST+":"+PORT+"/_bulk");
        } catch (MalformedURLException ex) {
            LOG.error("构造URL失败",ex);
        }
    }
    
    /**
     * 批量索引
     * 批量提交
     * 
     * @param <T> 泛型参数
     * @param list 批量模型
     */
    public <T extends Model> void index(List<T> list){
        success = 0;
        StringBuilder json = new StringBuilder();
        int j = 1;
        //构造批量索引请求
        for(T model : list){
            try{
                String simpleName = model.getClass().getSimpleName();
                LOG.debug((j++)+"、simpleName: 【"+simpleName+"】");
                json.append("{\"index\":{\"_index\":\"")
                        .append(INDEX_NAME)
                        .append("\",\"_type\":\"")
                        .append(simpleName)
                        .append("\"}}")
                        .append("\n");
                json.append("{");
                Field[] fields = model.getClass().getDeclaredFields();
                int len = fields.length;
                for(int i = 0; i < len; i++){
                    Field field = fields[i];
                    String name = field.getName();
                    field.setAccessible(true);
                    Object value = field.get(model);
                    //小心空指针异常，LogHandler线程会悄无声息地退出！
                    if(value == null){
                        LOG.debug("忽略空字段："+name);
                        continue;
                    }
                    if(i>0){
                        json.append(",");
                    }
                    String valueClass=value.getClass().getSimpleName();
                    LOG.debug("name: "+name+"   type: "+valueClass);
                    if("Timestamp".equals(valueClass) || "Date".equals(valueClass)){
                        //提交给ES的日期时间值要为"2014-01-31T13:53:54"这样的形式
                        value=DateTypeConverter.toDefaultDateTime((Date)value).replace(" ", "T");
                    }
                    String prefix = "\"";
                    String suffix = "\"";
                    //提交给ES的数字和布尔值不要加双引号
                    if("Float".equals(valueClass)
                            || "Double".equals(valueClass) 
                            || "Long".equals(valueClass) 
                            || "Integer".equals(valueClass)
                            || "Short".equals(valueClass)
                            || "Boolean".equals(valueClass)){
                        prefix="";
                        suffix="";
                    }
                    json.append("\"")
                            .append(name)
                            .append("\":")
                            .append(prefix)
                            .append(value)
                            .append(suffix);
                }
                json.append("}\n");
            }catch(SecurityException | IllegalArgumentException | IllegalAccessException e){
                LOG.error("构造索引请求失败【"+model.getMetaData()+"】\n"+model, e);
            }
        }
        //批量提交索引
        try{
            LOG.debug("提交JSON数据：\n"+json.toString());
            HttpURLConnection conn = (HttpURLConnection) URL.openConnection();
            conn.setRequestMethod("PUT");
            conn.setDoOutput(true);
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(conn.getOutputStream(),"utf-8"));    
            writer.write(json.toString());
            writer.flush();
            StringBuilder result = new StringBuilder();
            try (BufferedReader reader = new BufferedReader (new InputStreamReader (conn.getInputStream()))) {
                String line = reader.readLine();
                while(line != null){
                    result.append(line);
                    line = reader.readLine();
                }
            }
            String resultStr = result.toString();
            LOG.debug(resultStr);          
            //使用Jackson解析返回的JSON
            JsonNode node = MAPPER.readTree(resultStr);
            for(JsonNode item : node.get("items")){
                JsonNode createJsonNode = item.get("create");
                JsonNode okJsonNode = createJsonNode.get("ok");
                if(okJsonNode != null){
                    boolean r = okJsonNode.getBooleanValue();
                    if(r){
                        success++;
                    }
                }else{
                    JsonNode errorJsonNode = createJsonNode.get("error");
                    if(errorJsonNode != null){
                        String errorMessage = errorJsonNode.getTextValue();
                        LOG.error("索引失败："+errorMessage);
                    }
                }
            }
        }catch(IOException e){
            LOG.error("批量提交索引失败", e);
        }
    }
    
    @Override
    public <T extends Model> void handle(List<T> list) {
        LOG.info("开始将 "+list.size()+" 个日志对象索引到ElasticSearch服务器");
        long start = System.currentTimeMillis();
        index(list);
        long cost = System.currentTimeMillis() - start;
        
        if(success != list.size()){
            LOG.info("索引失败： "+(list.size()-success)+" 个");            
        }
        if(success > 0){
            LOG.info("索引成功： "+success+" 个");
        }
        LOG.info("耗时："+ConvertUtils.getTimeDes(cost));
    }
}

 

最后我们在配置文件config.local.properties中指定log.handlers的值为ElasticSearchLogHandler类的Spring bean name elasticSearchLogHandler，因为ElasticSearchLogHandler类加了Spring的@Service注解：

 

log.handlers=elasticSearchLogHandler

 

 

APDPlat托管在Github

分享到：
| APDPlat中备份文件异地容灾机制之FTP上传
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    maven springmvc集成elasticsearch客户端项目

    "es搜索引擎"标签进一步确认了我们正在处理一个与Elasticsearch相关的项目，Elasticsearch通常用于处理大规模数据的搜索和分析任务。 **文件名称列表详解：** 1. **.classpath** - 这是一个Eclipse项目文件，包含...

    最新版本springboot集成elasticsearch

    一般来说我们开发Elasticsearch会选择使用集成springboot，在网上找的springboot集成elasticsearch文章几乎都是extends ElasticsearchRepository的方式来实现。但是新版本ElasticsearchRepository里的方法基本上已经...

    Atlas2.2.0编译、安装及使用(集成ElasticSearch，导入Hive数据).doc

    Atlas2.2.0 可以集成 ElasticSearch，以提供更好的搜索功能。在集成 ElasticSearch 时，需要配置 ElasticSearch 的连接信息和索引信息。 导入 Hive 数据 Hive 是一个基于 Hadoop 的数据仓库工具，它提供了强大的...

    spring mvc 集成elasticSearch 5.5.0版本

    总的来说，通过Spring MVC与Elasticsearch的集成，我们可以轻松地在Web应用中利用Elasticsearch的强大搜索功能，实现高效的数据库操作。确保正确配置和理解Elasticsearch的基本概念，如索引、类型、映射等，是成功...

    springboot 集成elasticsearch ,api的使用

    在本文中，我们将深入探讨如何将Spring Boot与Elasticsearch集成，并使用其API进行数据操作。Elasticsearch是一个流行的开源搜索引擎，它提供了实时、分布式、容错能力的搜索和数据分析功能。Spring Boot则是一个...

    SpringBoot 集成ElasticSearch两个依赖的jar

    在Spring Boot应用中集成Elasticsearch是一项常见的任务，它能够帮助我们构建高性能的搜索和数据分析功能。Elasticsearch是一款开源的、分布式、实时的搜索与分析引擎，它被广泛用于日志分析、全文检索以及复杂的...

    elasticsearch sql支持增删改查且支持mybatis集成

    Elasticsearch SQL与MyBatis的集成意味着开发者可以在MyBatis的映射文件或注解中直接使用SQL查询Elasticsearch，使得Java应用能够无缝地与Elasticsearch交互，提高了开发效率和代码可读性。 4. **性能和效率**：...

    基于.netcore搜索封装ElasticSearch.zip

    这个名为"基于.netcore搜索封装ElasticSearch.zip"的压缩包，显然包含了一个针对.NET Core平台的Elasticsearch客户端库，方便开发者在.NET Core应用中集成和操作Elasticsearch。 Elasticsearch是一个开源的分布式...

    SpringBoot集成Elasticsearch（已实现各种ES操作，上手即可用）

    本实例属于简单易上手并且能够拿来就用的SpringBoot ES 项目，全文使用的是ElasticsearchTemplate进行开发。 本实例涵盖ES中的各类操作，如索引操作、CRUD操作、批处理、结果排序、分页查询、检索查询、关键字查询、...

    SpringBoot集成Elasticsearch（已实现各种ES操作，上手即可用

    本实例为博主原创，属于简单易上手并且能够拿来就用的SpringBoot ES 项目，全文使用的是ElasticsearchTemplate进行开发。 本实例涵盖ES中的各类操作，如索引操作、CRUD操作、批处理、结果排序、分页查询、检索查询、...

    基于Spring框架开源ElasticSearch搜索

    本书以实例讲述如何在Spring框架之上搭建ElasticSearch开发，以及如何利用JPA建立、更新和删除索引，如何配置ElasticSearch Server的applicationContext等。

    springboot集成elasticSearch.zip

    本示例"springboot集成elasticSearch.zip"提供了一个完整的集成方案，让我们详细探讨一下这个集成过程中的关键知识点。 首先，**集成准备**。在集成Spring Boot和Elasticsearch之前，我们需要确保已安装了Elastic...

    Springboot集成Elasticsearch+京东搜索实战代码

    在本项目中，"Springboot集成Elasticsearch+京东搜索实战代码" 主要涉及了两个核心技术：Spring Boot和Elasticsearch。Spring Boot是Spring框架的一个轻量级版本，旨在简化新Spring应用的初始搭建以及开发过程。而...

    springboot整合ElasticSearch集群

    springboot 2.0.2集成elasticsearch5.5.1，并使用集群模式，亲测可用！！！

    springMVC整合elasticsearch，基于maven

    SpringMVC作为Spring框架的一部分，主要用于构建Web应用程序的模型-视图-控制器（MVC）架构，而Elasticsearch则是一种分布式、RESTful风格的搜索和分析引擎，广泛应用于日志分析、实时监控、数据检索等领域。...

    Springboot集成elasticsearch

    Elasticsearch（简称ES）作为一个强大的分布式搜索引擎，因其易用性、高可扩展性和实时性能而备受青睐。Spring Boot，作为Java开发框架的明星产品，通过简化配置使得开发变得更加便捷。本文将深入探讨如何将Spring ...

    (狂神)ElasticSearch快速入门笔记，ElasticSearch基本操作以及爬虫(Java-ES仿京东实战)

    好记性不如烂笔头哦~，ElasticSearch，简称es，es是一个开源的高拓展的分布式全文搜索引擎它可以近乎实时的存储、检索数据；本身拓展性很好，可以拓展到上百台服务器，处理PB级别的数据。es也是用Java开发并使用...

    Elasticsearch集成Hadoop最佳实践

    2. 实时搜索与批处理分析：由于Elasticsearch提供了实时搜索的功能，而Hadoop擅长于批处理分析，因此集成后的系统可以实现实时的搜索需求，同时也不放弃Hadoop的批量分析能力。 3. 数据流处理：ES-Hadoop支持Kafka...

    php 的elasticsearch拓展

    PHP作为后端开发的主流语言之一，与Elasticsearch的集成是常见的需求。在PHP中，通过Elasticsearch的PHP客户端库，我们可以轻松地进行索引创建、文档增删改查以及复杂的数据分析操作。 本文将详细介绍如何在PHP中...

Magicbox
Global site tag (gtag.js) - Google Analytics