[大数据]ElasticSearch工具类

先把需要用的类注入进来

这两个类在meaven导入的时候就注入到spring了

	@Autowired
	private TransportClient transportClient;
	
	private static TransportClient client;
	

根据ID查找elasticsearch是否存在某条数据

 /**
     * 指定ID，查询数据
     * @param index
     * @param type
     * @param id
     * @param fields 需要显示的字段，逗号分隔（缺省为全部字段）
     * @return
     */
    public static Map<String,Object> searchDataById(String index, String type, String id, String fields) {
        GetRequestBuilder getRequestBuilder = client.prepareGet(index, type, id);
        if (StringUtils.isNotEmpty(fields)) {
            getRequestBuilder.setFetchSource(fields.split(","), null);
        }
        GetResponse getResponse = getRequestBuilder.execute().actionGet();
        return getResponse.getSource();
    }

判断索引是否存在

	/**
	 * 判断索引是否存在
	 * @param index
	 * @return
	 */
	public static boolean isIndexExist(String index) {
		IndicesExistsResponse inExistsResponse = client.admin().indices()
				.exists(new IndicesExistsRequest(index)).actionGet();
		if(inExistsResponse.isExists()) {
			System.out.println("索引 "+ index + " 已存在。");
		}
		else {
			System.out.println("索引 "+ index + " 不存在。");
		}
		return inExistsResponse.isExists();
	}
	

创建索引

	/**
	 * 创建索引
	 * @param index
	 * @return
	 */
    public static boolean createIndex(String index) {
    	if(!isIndexExist(index)) {
    		CreateIndexResponse indexResponse = client.admin().indices().prepareCreate(index).execute().actionGet();
            if(indexResponse.isAcknowledged()) {
            	System.out.println("索引 "+ index + " 创建成功。");
            }
            else {
            	System.out.println("索引 "+ index + " 创建失败。");
            }
            return indexResponse.isAcknowledged();
    	}
    	return false;
    }

删除索引

   /**
     * 删除索引
     * @param index
     * @return
     */
    public static boolean deleteIndex(String index) {
    	if(isIndexExist(index)) {
    		AcknowledgedResponse dResponse = client.admin().indices().prepareDelete(index).execute().actionGet();
            if (dResponse.isAcknowledged()) {
            	System.out.println("索引 "+ index + " 删除成功。");
            } else {
            	System.out.println("索引 "+ index + " 删除失败。");
            }
            return dResponse.isAcknowledged();
    	}
    	else {
    		return false;
    	}
    }

指定ID添加文档

    /**
     * 指定ID，添加文档(单条数据)
     * @param jsonObject
     * @param index
     * @param type
     * @param id
     * @return
     */
    public static String addData(JSONObject jsonObject, String index, String type, String id) {
        IndexResponse response = client.prepareIndex(index, type, id).setSource(jsonObject).get();
        System.out.println("添加数据状况: "+ response.status().getStatus() +" 。");
        return response.getId();
    }

其余的

   /**
     * 指定ID，删除文档(单条数据)
     * @param index
     * @param type
     * @param id
     */
    public static void deleteDataById(String index, String type, String id) {
        DeleteResponse response = client.prepareDelete(index, type, id).execute().actionGet();
        System.out.println("删除数据状况: "+ response.status().getStatus() +" 。");
    }
    
    /**
     * 指定ID，更新文档(单条数据)
     * @param jsonObject
     * @param index
     * @param type
     * @param id
     */
    public static void updateDataById(JSONObject jsonObject, String index, String type, String id) {
        UpdateRequest updateRequest = new UpdateRequest();
        updateRequest.index(index).type(type).id(id).doc(jsonObject);
        client.update(updateRequest);
    }

    /**
     * 指定ID，查询数据
     * @param index
     * @param type
     * @param id
     * @param fields 需要显示的字段，逗号分隔（缺省为全部字段）
     * @return
     */
    public static Map<String,Object> searchDataById(String index, String type, String id, String fields) {
        GetRequestBuilder getRequestBuilder = client.prepareGet(index, type, id);
        if (StringUtils.isNotEmpty(fields)) {
            getRequestBuilder.setFetchSource(fields.split(","), null);
        }
        GetResponse getResponse = getRequestBuilder.execute().actionGet();
        return getResponse.getSource();
    }
    
    /**
     * 分词查询，并高亮
     *
     * @param index          
     * @param type          
     * @param query          查询条件
     * @param size           文档大小限制
     * @param fields         需要显示的字段，逗号分隔（缺省为全部字段）
     * @param sortField      排序字段
     * @param highlightField 高亮字段
     * @return
     */
    public static Map<String, Object> searchListData(String index, String type, QueryBuilder query, Integer size, String fields, String sortField, String highlightField) {

        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index);
        if (StringUtils.isNotEmpty(type)) {
            searchRequestBuilder.setTypes(type.split(","));
        }

        //设置查询条件
        searchRequestBuilder.setQuery(query);

        //需要显示的字段，逗号分隔（缺省为全部字段）
        if (StringUtils.isNotEmpty(fields)) {
            searchRequestBuilder.setFetchSource(fields.split(","), null);
        }
        searchRequestBuilder.setFetchSource(true);
        
        //设置是否按查询匹配度排序
        searchRequestBuilder.setExplain(true);

        //排序字段
//        if (StringUtils.isNotEmpty(sortField)) {
//            searchRequestBuilder.addSort(sortField, SortOrder.DESC);
//        }
        
        //设置高亮
        if (StringUtils.isNotEmpty(highlightField)) {
            HighlightBuilder highlightBuilder = new HighlightBuilder(); 
            
            String[] highlightFields = highlightField.split(",");
            for(String hField: highlightFields) {
            	highlightBuilder.field(new HighlightBuilder.Field(hField));
            }
            
            highlightBuilder.preTags("<span style='color:red'>");//设置前缀
            highlightBuilder.postTags("</span>");//设置后缀
            
            highlightBuilder.fragmentSize(800000); //最大高亮分片数
    		highlightBuilder.numOfFragments(0);
    		
            searchRequestBuilder.highlighter(highlightBuilder);
        }

        if (size != null && size > 0) {
            searchRequestBuilder.setSize(size);
        }

        //打印的内容,可以在 Elasticsearch head 和 Kibana 上执行查询
        System.out.println(searchRequestBuilder);

        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        
        long total = searchResponse.getHits().totalHits;
        long length = searchResponse.getHits().getHits().length;
        
        System.out.println("共查询到" + total + "条数据,处理数据条数为" + length +"条。");

        if (searchResponse.status().getStatus() == 200) {
        	
//        	List<Map<String, Object>> sourceList = new ArrayList<Map<String, Object>>();
//        	for (SearchHit searchHit : searchResponse.getHits().getHits()) {
//        		sourceList.add(searchHit.getSourceAsMap());
//        	}
//        	return sourceList;
        	
        	//实现高亮
        	List<Map<String, Object>> sourceList = setSearchResponse(searchResponse, highlightField);
        	
        	Map<String, Object> map = new HashMap<String, Object>();        	
        	map.put("total", total);
        	map.put("sourceList", sourceList);
        	return map;
        }

        return null;
    }
    
    /**
     * 分词查询，并分页、高亮
     *
     * @param index          
     * @param type 
     * @param startPage      当前页
     * @param pageSize       每页显示条数
     * @param query          查询条件
     * @param fields         需要显示的字段，逗号分隔（缺省为全部字段）
     * @param sortField      排序字段
     * @param highlightField 高亮字段
     * @return
     */
    public static Map<String, Object> searchListDataPage(String index, String type, int startPage, int pageSize, QueryBuilder query, String fields, String sortField, String highlightField) {

        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index);
        if (StringUtils.isNotEmpty(type)) {
            searchRequestBuilder.setTypes(type.split(","));
        }

        searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH);         

        //需要显示的字段，逗号分隔（缺省为全部字段）
        if (StringUtils.isNotEmpty(fields)) {
            searchRequestBuilder.setFetchSource(fields.split(","), null);
        }
        
        //排序字段
        if (StringUtils.isNotEmpty(sortField)) {
            searchRequestBuilder.addSort(sortField, SortOrder.DESC);
        }
        
        //设置高亮
        if (StringUtils.isNotEmpty(highlightField)) {
            HighlightBuilder highlightBuilder = new HighlightBuilder(); 
            
            String[] highlightFields = highlightField.split(",");
            for(String hField: highlightFields) {
            	highlightBuilder.field(new HighlightBuilder.Field(hField));
            }
            
            // 设置高亮字段
//            HighlightBuilder.Field highlightTitle = new HighlightBuilder.Field("title");
//            highlightBuilder.field(highlightTitle);
//            
//            HighlightBuilder.Field highlightDetail = new HighlightBuilder.Field("details");
//            highlightBuilder.field(highlightDetail);            
            
            highlightBuilder.preTags("<span style='color:red'>");//设置前缀
            highlightBuilder.postTags("</span>");//设置后缀
            
            highlightBuilder.fragmentSize(800000); //最大高亮分片数
    		highlightBuilder.numOfFragments(0);
    		
            searchRequestBuilder.highlighter(highlightBuilder);
        }
       
        //设置查询条件
        searchRequestBuilder.setQuery(query);
        
        //分页应用
        searchRequestBuilder.setFrom((startPage-1)*pageSize).setSize(pageSize);

        //设置是否按查询匹配度排序
        searchRequestBuilder.setExplain(true);         

        //打印的内容,可以在 Elasticsearch head 和 Kibana 上执行查询
        System.out.println(searchRequestBuilder);

        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); 
        
        long total = searchResponse.getHits().totalHits;
        long length = searchResponse.getHits().getHits().length;
        
        System.out.println("共查询到" + total + "条数据,处理数据条数为" + length +"条。");
        
        if (searchResponse.status().getStatus() == 200) {
//        	List<Map<String, Object>> sourceList = new ArrayList<Map<String, Object>>();
//        	for (SearchHit searchHit : searchResponse.getHits().getHits()) {
//        		sourceList.add(searchHit.getSourceAsMap());
//        	}
        	
        	//实现高亮
        	List<Map<String, Object>> sourceList = setSearchResponse(searchResponse, highlightField);
        	
        	Map<String, Object> map = new HashMap<String, Object>();        	
        	map.put("total", total);
        	map.put("sourceList", sourceList);
        	return map;
        }

        return null;
    }
    
    /**
     * 批量导入所有文档（不是部分更新），该代码不能共用，只适用于“ESKnowledgeEntity”
     * 先删除索引（原来所有数据），再添加数据
     * @param know_list
     * @param index
     * @param type
     */
    public static void addAllData(List<ESKnowledgeEntity> know_list, String index, String type) {
    	
    	//删除索引（删除索引的时候，会去判断该索引是否存在）
    	deleteIndex(index);

    	
    	//detail内容有大量的html标签，去除html标签
    	know_list.forEach(esKnowledgeEntity -> {
            esKnowledgeEntity.setDetails(delHTMLTag(esKnowledgeEntity.getDetails()));
        });
    	
        //建立批量提交类
        BulkRequestBuilder bulkRequest= client.prepareBulk();
        for(ESKnowledgeEntity know: know_list) {
        	//将实体类直接转为JSON字符串
        	String source = JSONSerializer.toJSON(know).toString();
        	bulkRequest.add(client.prepareIndex(index,type,know.getRecId()).setSource(source,XContentType.JSON)); 
        }
        bulkRequest.execute().actionGet();
        
        System.out.println("共查询到" + know_list.size() + "条数据,已导入成功");
    }
    
    
    /**
     * 批量导入部分文档，该代码不能共用，只适用于“ESKnowledgeEntity”
     * 无需删除原有文档，已存在的文档将更新，不存在的文档选择插入
     * @param know_list
     * @param index
     * @param type
     */
    public static boolean addListData(List<ESKnowledgeEntity> know_list, String index, String type) {
    	
    	//和以上方法同样，该导入方法也必须判断当前索引是否存在
    	boolean flag = false;
    	int count = 0;
    	
    	if(isIndexExist(index)) {
    		//detail内容有大量的html标签，去除html标签
        	know_list.forEach(esKnowledgeEntity -> {
                esKnowledgeEntity.setDetails(delHTMLTag(esKnowledgeEntity.getDetails()));
            });
        	
        	for(ESKnowledgeEntity know: know_list) {
            	//将实体类直接转为JSON字符串,再转为JSONObject
            	String source = JSONSerializer.toJSON(know).toString();
            	JSONObject json = JSONObject.parseObject(source);
            	
            	//查询相应ID的数据，判断是否已存在
            	Map<String,Object> map = searchDataById(index, type, know.getRecId(), null);
            	//存在就更新，不存在就插入
            	if(map!=null) {
            		updateDataById(json, index, type, know.getRecId());
            	}
            	else {
            		addData(json, index, type, know.getRecId());
            	}
            	count++;
            }       	
        	System.out.println("共查询到" + count + "条数据,已导入成功");
        	flag = true;
        	return flag;
    	}
    	
    	return flag;
    }
    
    
    /**
     * 去除details中html标签
     * @param htmlStr
     * @return
     */
    public static String delHTMLTag(String htmlStr) {
        String regEx_style = "<style[^>]*?>[\\s\\S]*?<\\/style>"; //定义style的正则表达式
        String regEx_html = "<[^>]+>"; //定义HTML标签的正则表达式

        Pattern p_style = Pattern.compile(regEx_style, Pattern.CASE_INSENSITIVE);
        Matcher m_style = p_style.matcher(htmlStr);
        htmlStr = m_style.replaceAll(""); //过滤style标签

        Pattern p_html = Pattern.compile(regEx_html, Pattern.CASE_INSENSITIVE);
        Matcher m_html = p_html.matcher(htmlStr);
        htmlStr = m_html.replaceAll(""); //过滤html标签

//        htmlStr = htmlStr.replace(" ", "");
//        htmlStr = htmlStr.replaceAll("\\s*|\t|\r|\n", "");
//        htmlStr = htmlStr.replace("“", "");
//        htmlStr = htmlStr.replace("”", "");
//        htmlStr = htmlStr.replaceAll("　", "");
        htmlStr = htmlStr.replace("&nbsp;", "");

        return htmlStr.trim(); //返回文本字符串
    }
    
    /**
     * 高亮处理
     * @param searchResponse
     * @param highlightField
     * @return
     */
    private static List<Map<String, Object>> setSearchResponse(SearchResponse searchResponse, String highlightField) {
        List<Map<String, Object>> sourceList = new ArrayList<Map<String, Object>>();
        StringBuffer stringBuffer = new StringBuffer();
        String[] hFields = highlightField.split(",");
        for (SearchHit searchHit : searchResponse.getHits().getHits()) {
            searchHit.getSourceAsMap().put("id", searchHit.getId());
            for(String hFiled:hFields) {
            	highlightField = hFiled;
            	if (StringUtils.isNotEmpty(highlightField)) {
                	stringBuffer = new StringBuffer();
                    //System.out.println("遍历 高亮结果集，覆盖 正常结果集" + searchHit.getSourceAsMap());
                	
                	Map<String, HighlightField> highlightFields = searchHit.getHighlightFields();
                	
                	if(highlightFields.containsKey(highlightField)) {
                		Text[] text = searchHit.getHighlightFields().get(highlightField).getFragments();

                        if (text != null) {
                            for (Text str : text) {
                                stringBuffer.append(str.string());
                            }
                            //遍历 高亮结果集，覆盖 正常结果集
                            searchHit.getSourceAsMap().put(highlightField, stringBuffer.toString());
                        }
                	}              
                }
            }          
            sourceList.add(searchHit.getSourceAsMap());
        }

        return sourceList;
    }