基于员工信息的CRUD操作
/** * 员工增删改查的应用程序 * * @author Administrator * */public class EmployeeCRUDApp { @SuppressWarnings({ "unchecked", "resource" }) public static void main(String[] args) throws Exception { // 先构建client Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build(); TransportClient client = new PreBuiltTransportClient(settings) .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300)); // createEmployee(client); // getEmployee(client); // updateEmployee(client); // deleteEmployee(client); client.close(); } /** * 创建员工信息（创建一个document） * * @param client */ private static void createEmployee(TransportClient client) throws Exception { IndexResponse response = client.prepareIndex("company", "employee", "1") .setSource(XContentFactory.jsonBuilder().startObject().field("name", "jack").field("age", 27) .field("position", "technique").field("country", "china").field("join_date", "2017-01-01") .field("salary", 10000).endObject()) .get(); System.out.println(response.getResult()); } /** * 获取员工信息 * * @param client * @throws Exception */ private static void getEmployee(TransportClient client) throws Exception { GetResponse response = client.prepareGet("company", "employee", "1").get(); System.out.println(response.getSourceAsString()); } /** * 修改员工信息 * * @param client * @throws Exception */ private static void updateEmployee(TransportClient client) throws Exception { UpdateResponse response = client.prepareUpdate("company", "employee", "1") .setDoc(XContentFactory.jsonBuilder().startObject().field("position", "technique manager").endObject()) .get(); System.out.println(response.getResult()); } /** * 删除 员工信息 * * @param client * @throws Exception */ private static void deleteEmployee(TransportClient client) throws Exception { DeleteResponse response = client.prepareDelete("company", "employee", "1").get(); System.out.println(response.getResult()); }} 基于员工信息的查询操作
/** * 员工搜索应用程序 * * @author Administrator * */public class EmployeeSearchApp { @SuppressWarnings({ "unchecked", "resource" }) public static void main(String[] args) throws Exception { Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build(); TransportClient client = new PreBuiltTransportClient(settings) .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300)); prepareData(client); // executeSearch(client); client.close(); } /** * 执行搜索操作 * * @param client */ private static void executeSearch(TransportClient client) { SearchResponse response = client.prepareSearch("company").setTypes("employee") .setQuery(QueryBuilders.matchQuery("position", "technique")) .setPostFilter(QueryBuilders.rangeQuery("age").from(30).to(40)).setFrom(0).setSize(1).get(); SearchHit[] searchHits = response.getHits().getHits(); for (int i = 0; i < searchHits.length; i++) { System.out.println(searchHits[i].getSourceAsString()); } } /** * 准备数据 * * @param client */ private static void prepareData(TransportClient client) throws Exception { client.prepareIndex("company", "employee", "1") .setSource(XContentFactory.jsonBuilder().startObject().field("name", "jack").field("age", 27) .field("position", "technique software").field("country", "china") .field("join_date", "2017-01-01").field("salary", 10000).endObject()) .get(); client.prepareIndex("company", "employee", "2") .setSource(XContentFactory.jsonBuilder().startObject().field("name", "marry").field("age", 35) .field("position", "technique manager").field("country", "china") .field("join_date", "2017-01-01").field("salary", 12000).endObject()) .get(); client.prepareIndex("company", "employee", "3") .setSource(XContentFactory.jsonBuilder().startObject().field("name", "tom").field("age", 32) .field("position", "senior technique software").field("country", "china") .field("join_date", "2016-01-01").field("salary", 11000).endObject()) .get(); client.prepareIndex("company", "employee", "4") .setSource(XContentFactory.jsonBuilder().startObject().field("name", "jen").field("age", 25) .field("position", "junior finance").field("country", "usa").field("join_date", "2016-01-01") .field("salary", 7000).endObject()) .get(); client.prepareIndex("company", "employee", "5") .setSource(XContentFactory.jsonBuilder().startObject().field("name", "mike").field("age", 37) .field("position", "finance manager").field("country", "usa").field("join_date", "2015-01-01") .field("salary", 15000).endObject()) .get(); }} 基于员工新的聚合查询操作
/** * 员工聚合分析应用程序 * * @author Administrator * */public class EmployeeAggrApp { @SuppressWarnings({ "unchecked", "resource" }) public static void main(String[] args) throws Exception { Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build(); TransportClient client = new PreBuiltTransportClient(settings) .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300)); SearchResponse searchResponse = client.prepareSearch("company") .addAggregation(AggregationBuilders.terms("group_by_country").field("country") .subAggregation(AggregationBuilders.dateHistogram("group_by_join_date").field("join_date") .dateHistogramInterval(DateHistogramInterval.YEAR) .subAggregation(AggregationBuilders.avg("avg_salary").field("salary")))) .execute().actionGet(); Map aggrMap = searchResponse.getAggregations().asMap(); StringTerms groupByCountry = (StringTerms) aggrMap.get("group_by_country"); Iterator groupByCountryBucketIterator = groupByCountry.getBuckets().iterator(); while (groupByCountryBucketIterator.hasNext()) { Bucket groupByCountryBucket = groupByCountryBucketIterator.next(); System.out.println(groupByCountryBucket.getKey() + ":" + groupByCountryBucket.getDocCount()); Histogram groupByJoinDate = (Histogram) groupByCountryBucket.getAggregations().asMap() .get("group_by_join_date"); Iterator groupByJoinDateBucketIterator = groupByJoinDate .getBuckets().iterator(); while (groupByJoinDateBucketIterator.hasNext()) { org.elasticsearch.search.aggregations.bucket.histogram.Histogram.Bucket groupByJoinDateBucket = groupByJoinDateBucketIterator .next(); System.out.println(groupByJoinDateBucket.getKey() + ":" + groupByJoinDateBucket.getDocCount()); Avg avg = (Avg) groupByJoinDateBucket.getAggregations().asMap().get("avg_salary"); System.out.println(avg.getValue()); } } client.close(); }} 注：聚合查询的时候可能出现问题：fielddata需要变为true，这个时候需要手动添加mapping
GET /company/_mapping/employee { "company": { "mappings": { "employee": { "properties": { "age": { "type": "long" }, "country": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } }, "join_date": { "type": "date" }, "name": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } }, "position": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } }, "salary": { "type": "long" } } } } }}Delete /company上面查出来以后进行截取，修改"fielddata": truePUT /company{ "mappings": { "employee": { "properties": { "age": { "type": "long" }, "country": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } }, "fielddata": true }, "join_date": { "type": "date" }, "name": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } }, "position": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } }, "salary": { "type": "long" } } } }} elasticsearch-5.2.0获取代码：
import org.apache.commons.lang.StringUtils;import org.elasticsearch.client.Client;import org.elasticsearch.client.transport.TransportClient;import org.elasticsearch.common.settings.Settings;import org.elasticsearch.common.transport.TransportAddress;import org.elasticsearch.transport.client.PreBuiltTransportClient;import org.springframework.stereotype.Component;import com.ad.utils.ConfigUtil;import javax.annotation.PostConstruct;import javax.annotation.Resource;import java.net.InetAddress;import java.util.ArrayList;import java.util.List;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;/** * Created by wangyunpeng on 2017/8/5. */@Componentpublic class ElasticSearchClient { @Resource(type=ConfigUtil.class) private ConfigUtil configUtil; private Map clientMap = new ConcurrentHashMap (); /*1.@PostConstruct说明 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器调用一次，类似于Serclet的inti()方法。被@PostConstruct修饰的方法会在构造函数之后，init()方法之前运行。2.@PreConstruct说明 被@PreConstruct修饰的方法会在服务器卸载Servlet的时候运行，并且只会被服务器调用一次，类似于Servlet的destroy()方法。被@PreConstruct修饰的方法会在destroy()方法之后运行，在Servlet被彻底卸载之前。 */ @PostConstruct public void init() { init(configUtil.clusterName, configUtil.clusterIpAddress); //init(configUtil.clusterName2, configUtil.clusterIpAddress2); } private void init(String clusterName, String clusterIpAddress){ try { Settings settings = Settings.builder() .put("cluster.name", clusterName) .build(); addClient(settings, getAllAddress(clusterIpAddress)); }catch(Exception e){ e.printStackTrace(); } } /** * 得所有的地址端口 * @param ips * @return * @throws Exception */ public List getAllAddress(String ips)throws Exception { List addressList = new ArrayList (); if(StringUtils.isNotBlank(ips)&&ips.contains(",")){ String[] ipaddr=ips.split(","); for (int i=0;i transportAddress) throws Exception{ TransportClient client = new PreBuiltTransportClient(setting); for(int i=0;i elasticsearch-6.0.0获取代码：
import org.apache.commons.lang.StringUtils;import org.elasticsearch.client.Client;import org.elasticsearch.client.transport.TransportClient;import org.elasticsearch.common.settings.Settings;import org.elasticsearch.common.transport.InetSocketTransportAddress;import org.elasticsearch.transport.client.PreBuiltTransportClient;import org.springframework.stereotype.Component;import com.ad.utils.ConfigUtil;import javax.annotation.PostConstruct;import javax.annotation.Resource;import java.net.InetAddress;import java.util.ArrayList;import java.util.List;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;/** * Created by wangyunpeng on 2017/8/5. */@Componentpublic class ElasticSearchClient { @Resource(type=ConfigUtil.class) private ConfigUtil configUtil; private Map clientMap = new ConcurrentHashMap (); /*1.@PostConstruct说明 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器调用一次，类似于Serclet的inti()方法。被@PostConstruct修饰的方法会在构造函数之后，init()方法之前运行。2.@PreConstruct说明 被@PreConstruct修饰的方法会在服务器卸载Servlet的时候运行，并且只会被服务器调用一次，类似于Servlet的destroy()方法。被@PreConstruct修饰的方法会在destroy()方法之后运行，在Servlet被彻底卸载之前。 */ @PostConstruct public void init() { init(configUtil.clusterName, configUtil.clusterIpAddress); //init(configUtil.clusterName2, configUtil.clusterIpAddress2); } private void init(String clusterName, String clusterIpAddress){ try { Settings settings = Settings.builder() .put("cluster.name", clusterName) .build(); addClient(settings, getAllAddress(clusterIpAddress)); }catch(Exception e){ e.printStackTrace(); } /*try { Settings settings = TransportClient.builder().build().settings().builder().put("cluster.name", clusterName).build(); addClient(settings, getAllAddress(clusterIpAddress)); }catch(Exception e){ e.printStackTrace(); }*/ } /** * 得所有的地址端口 * @param ips * @return * @throws Exception */ public List getAllAddress(String ips)throws Exception { List addressList = new ArrayList (); if(StringUtils.isNotBlank(ips)&&ips.contains(",")){ String[] ipaddr=ips.split(","); for (int i=0;i transportAddress) throws Exception{ TransportClient client = new PreBuiltTransportClient(setting); //TransportClient client = TransportClient.builder().settings(setting).build(); for(int i=0;i maven依赖：
org.elasticsearch.client transport 6.0.0
测试代码：
import java.io.IOException;import java.util.UUID;import javax.annotation.Resource;import org.elasticsearch.action.index.IndexResponse;import org.elasticsearch.client.Client;import org.elasticsearch.common.xcontent.XContentFactory;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;import com.ad.base.AlbumIndexField;import com.ad.utils.ConfigUtil;import com.ad.utils.ExceptionUtil;import net.sf.json.JSONObject;@Componentpublic class ElasticSaveServiceImpl implements ElasticSaveService { private static final Logger logger = LoggerFactory.getLogger(ElasticSearchServiceImpl.class); @Resource(type = ElasticSearchClient.class) private ElasticSearchClient elasticSearchClient; @Resource(type = ConfigUtil.class) private ConfigUtil configUtil; /** * @return */ public Client getClient(String clusterName) { try { return elasticSearchClient.getClient(clusterName); } catch (Exception e) { logger.error("ES获取client失败 :" + ExceptionUtil.stackTrace(e)); return null; } } @Override public void executeSave(String clusterName, String json1) { JSONObject json = JSONObject.fromObject(json1); String documentId = (UUID.randomUUID().toString().replaceAll("-", "")); IndexResponse response; try { response = this.getClient(clusterName).prepareIndex(configUtil.indexName, configUtil.indexType, documentId) .setSource(XContentFactory.jsonBuilder() .startObject() .field(AlbumIndexField.FID, json.getString("fid")) .endObject()) .get(); } catch (IOException e) { logger.error("===AdInfoConsumer consumer is exception", e); //e.printStackTrace(); } }}