JAVA API操作ElasticSearch
15886 ワード
org.springframework.boot
spring-boot-starter-parent
2.0.1.RELEASE
com.dtmall.es
demo
1.0.0-SNAPSHOT
jar
es_demo
http://maven.apache.org
UTF-8
5.6.4
1.8
spring-libs-milestone
https://repo.spring.io/libs-milestone
spring-plugins-release
https://repo.spring.io/plugins-release
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-test
test
org.elasticsearch
elasticsearch
${elasticsearch.version}
org.elasticsearch.client
transport
${elasticsearch.version}
commons-logging
commons-logging
/***
** ES,
*/
@Configuration
public class EsConfig{
@Value("${elasticsearch.host}")
private String esHost;
@Value("${elasticsearch.port}")
private int esPort;
@Value("${elasticsearch.clusterName}")
private String esClusterName;
private TransportClient client;
@PostConstruct
public void initialize() throws Exception {
Settings esSettings = Settings.builder()
.put("cluster.name", esClusterName)
.put("client.transport.sniff", true).build();
client = new PreBuiltTransportClient(esSettings);
String[] esHosts = esHost.trim().split(",");
for (String host : esHosts) {
client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host),
esPort));
}
}
@Bean
public Client client() {
return client;
}
@PreDestroy
public void destroy() {
if (client != null) {
client.close();
}
}
}
@Component
public class ElasticOperationService {
private final Logger logger = LoggerFactory.getLogger(ElasticOperationService.class);
@Autowired
private Client client;
private BulkProcessor bulkProcessor;
@PostConstruct
public void initBulkProcessor() {
bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
logger.info(" ：{} {} ",executionId,request.numberOfActions());
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
logger.error(String.format(" ：%s ; ：%s",executionId,request.numberOfActions()),failure);
}
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
logger.info(" ：{} {} , ：{} ,",executionId,request.numberOfActions(),response.getTookInMillis());
}
}).setBulkActions(1000)
.setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB))
.setConcurrentRequests(4)
.setFlushInterval(TimeValue.timeValueSeconds(5))
.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(500),3)) //
.build();
}
@PreDestroy
public void closeBulk() {
if(bulkProcessor != null) {
try {
bulkProcessor.close();
}catch (Exception e) {
logger.error("close bulkProcessor exception",e);
}
}
}
/**
* ,
*
*/
public void addDocumentToBulkProcessor(String indices, String type, Object object) {
bulkProcessor.add(client.prepareIndex(indices, type).setSource(JsonUtils.beanToJson(object)).request());
}
public void addDocument(String indices, String type, Object object) {
IndexResponse resp = client.prepareIndex(indices, type).setSource(JsonUtils.beanToJson(object)).get();
logger.info(" ：{}",resp.toString());
}
/**
* id
*
*/
public void deleteDocumentById(String index, String type, String id) {
// new DeleteByQueryRequest(search);
DeleteResponse resp = client.prepareDelete(index, type, id).get();
logger.info(" ：{}",resp.toString());
}
/**
*
*
*/
public void deleteDocumentByQuery(String index, String type, UserSearchParam param) {
//DeleteByQueryRequestBuilder builder = new DeleteByQueryRequestBuilder(client,DeleteByQueryAction.INSTANCE);
DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE.newRequestBuilder(client);
//builder.filter(convertParam(param));
builder.source().setIndices(index).setTypes(type).setQuery(convertParam(param));
BulkByScrollResponse resp = builder.get();
logger.info(" ：{}",resp.toString());
}
/**
* ID
*
*/
public void updateDocument(String indices, String type,String id,Object object) {
UpdateResponse resp = client.prepareUpdate(indices, type, id).setDoc(JsonUtils.beanToJson(object)).get();
logger.info(" ：{}",resp.toString());
}
/**
*
*
*/
public void updateDocumentByQuery(String indices, String type, Object object,UserSearchParam param) {
//UpdateByQueryRequestBuilder builder = new UpdateByQueryRequestBuilder(client,UpdateByQueryAction.INSTANCE);
UpdateByQueryRequestBuilder builder = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
builder.source().setIndices(indices).setTypes(type).setQuery(convertParam(param));
}
public List queryDocumentByParam(String indices, String type,UserSearchParam param,Class clazz) {
SearchRequestBuilder builder = buildRequest(indices,type);
builder.addSort("birthday",SortOrder.ASC);
builder.setQuery(convertParam(param));
builder.setFrom(0).setSize(10);
SearchResponse resp = builder.get();
return convertResponse(resp,clazz);
}
private BoolQueryBuilder convertParam(UserSearchParam param) {
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
if (StringUtils.hasText(param.getUserName())) {
boolQueryBuilder.must(QueryBuilders.termQuery("userName", param.getUserName()));
}
if (param.getAge() != null) {
boolQueryBuilder.must(QueryBuilders.rangeQuery("age").gt(param.getAge()));
}
if (StringUtils.hasText(param.getDescription())) {
boolQueryBuilder.must(QueryBuilders.matchQuery("description", param.getDescription()));
}
if(StringUtils.hasText(param.getRoleName())) {
boolQueryBuilder.must(QueryBuilders.nestedQuery("roles", QueryBuilders.termQuery("roles.name", param.getRoleName()), ScoreMode.None));
}
return boolQueryBuilder;
}
/**
*
*
*/
public List convertResponse(SearchResponse response,Class clazz) {
List list = Lists.newArrayList();
if(response != null && response.getHits() != null) {
String result = org.apache.commons.lang3.StringUtils.EMPTY;
T e = null;
Field idField = ReflectionUtils.findField(clazz, "id");
if (idField != null) {
ReflectionUtils.makeAccessible(idField);
}
for(SearchHit hit : response.getHits()) {
result = hit.getSourceAsString();
if (StringUtils.hasText(result)) {
e = JsonUtils.jsonToBean(result, clazz);
}
if (e != null) {
if (idField != null) {
ReflectionUtils.setField(idField, e, hit.getId());
}
list.add(e);
}
}
}
return list;
}
public SearchRequestBuilder buildRequest(String indices, String type) {
return client.prepareSearch(indices).setTypes(type);
}
/**
*
*
*/
public boolean createIndexIfNotExist(String index, String type) {
IndicesAdminClient adminClient = client.admin().indices();
IndicesExistsRequest request = new IndicesExistsRequest(index);
IndicesExistsResponse response = adminClient.exists(request).actionGet();
if (!response.isExists()) {
return createIndex(index, type);
}
return true;
}
/**
*
*
*/
public boolean createIndex(String index, String type) {
XContentBuilder mappingBuilder;
try {
mappingBuilder = this.getMapping(type);
} catch (Exception e) {
logger.error(String.format(" Mapping ；index:%s type:%s,", index, type), e);
return false;
}
Settings settings = Settings.builder().put("index.number_of_shards", 2)
.put("index.number_of_replicas", 1)
.put("index.refresh_interval", "5s").build();
IndicesAdminClient adminClient = client.admin().indices();
CreateIndexRequestBuilder builder = adminClient.prepareCreate(index);
builder.setSettings(settings);
CreateIndexResponse response = builder.addMapping(type, mappingBuilder).get();
logger.info(" ：{} :{} ：{}", index, type, response.isAcknowledged());
return response.isAcknowledged();
}
/***
* Mapping roles nested
*/
private XContentBuilder getMapping(String type) throws Exception {
XContentBuilder mappingBuilder = XContentFactory.jsonBuilder().startObject().startObject(type)
.startObject("_all").field("enabled", false).endObject()
.startObject("properties")
.startObject("userName").field("type", "keyword").endObject()
.startObject("age").field("type", "integer").endObject()
.startObject("birthday").field("type", "date").endObject()
.startObject("description").field("type", "text").field("analyzer", "ik_smart").endObject()
.startObject("roles").field("type", "nested")
.startObject("properties")
.startObject("createTime").field("type","date").endObject()
.startObject("name").field("type","keyword").endObject()
.startObject("description").field("type","text").field("analyzer", "ik_smart").endObject()
.endObject()
.endObject()
.endObject()
.endObject().endObject();
return mappingBuilder;
}
}
/**
* USER ， 1 roles
**/
public class User {
private String id;
private String userName;
private Integer age;
private Date birthday;
private String description;
/**
* 1 spring-data-elasticsearch nested
**/
private List roles;
public User() {}
public User(String userName,Integer age,Date birthday,String description) {
this.userName = userName;
this.age = age;
this.birthday = birthday;
this.description = description;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getUserName() {
return userName;
}
public void setUserName(String userName) {
this.userName = userName;
}
public Date getBirthday() {
return birthday;
}
public void setBirthday(Date birthday) {
this.birthday = birthday;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public List getRoles() {
return roles;
}
public void setRoles(List roles) {
this.roles = roles;
}
}
public class Role {
private String id;
private String name;
private Date createTime;
private String description;
public Role() {}
public Role(String name,String description,Date createTime) {
this.name = name;
this.description = description;
this.createTime = createTime;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Date getCreateTime() {
return createTime;
}
public void setCreateTime(Date createTime) {
this.createTime = createTime;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
}
@Service
public class OrgUserService implements StandardService,InitializingBean{
@Autowired
private ElasticOperationService elasticOperationService;
private String index = "test-api";
private String type = "user";
@Override
public void afterPropertiesSet() throws Exception {
elasticOperationService.createIndexIfNotExist(index, type);
}
public void batchAddUser(List users) {
if(CollectionUtils.isEmpty(users)) {
return ;
}
for(User user :users) {
elasticOperationService.addDocumentToBulkProcessor(index, type, user);
}
}
@Override
public void addUser(User user) {
elasticOperationService.addDocument(index, type, user);
}
@Override
public void deletedUserById(String id) {
elasticOperationService.deleteDocumentById(index, type, id);
}
@Override
public void updateUser(User user) {
String id = user.getId();
user.setId(null);
elasticOperationService.updateDocument(index, type,id, user);
}
@Override
public List queryByUserName(String userName) {
UserSearchParam param = new UserSearchParam();
param.setUserName(userName);
return elasticOperationService.queryDocumentByParam(index, type, param,User.class);
}
@Override
public List queryByRoleName(Role role) {
UserSearchParam param = new UserSearchParam();
param.setRoleName(role.getName());
return elasticOperationService.queryDocumentByParam(index, type, param,User.class);
}
}