- 引入依賴
- 初始化RestHighLevelClient和BulkProcessor對象
- 增刪改操作
3.1 數(shù)據(jù)準(zhǔn)備
3.2 單條數(shù)據(jù)異步插入
3.3 單條數(shù)據(jù)同步插入
3.4 批量插入
3.5 更新操作
3.6 帶條件的更新語句
3.7 批量更新
3.8 刪除操作
3.9 條件刪除
Java層面操作elasticSearch7.x以清，為了便于操作饱狂，不集成Spring嗦锐，使用main方法進(jìn)行調(diào)用淹朋。
1. 引入依賴
<!--解決：java.lang.NoClassDefFoundError: org/elasticsearch/common/xcontent/DeprecationHandler-->
<!-- elasticsearch -->
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>7.5.1</version>
</dependency>
<!-- elasticsearch-rest-client -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
<version>7.5.1</version>
</dependency>
<!-- elasticsearch-rest-high-level-client -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.5.1</version>
<exclusions>
<exclusion>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
</exclusion>
<exclusion>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
</exclusion>
</exclusions>
</dependency>
2. 初始化RestHighLevelClient和BulkProcessor對象
RestHighLevelClientRestHighLevelClient是官方指定的連接API。另外一個是TransportClient而叼，但是TransportClient這個是已經(jīng)廢棄不用的塔粒，所以會在ES8.0之后完全移除品追，也就是說8.0之后就無法使用了。
@Slf4j
public class EsTest {
//es操作客戶端
private static RestHighLevelClient restHighLevelClient;
//批量操作的對象
private static BulkProcessor bulkProcessor;
static {
List<HttpHost> httpHosts = new ArrayList<>();
//填充數(shù)據(jù)
httpHosts.add(new HttpHost("172.26.17.11", 9200));
httpHosts.add(new HttpHost("172.26.17.11", 9201));
httpHosts.add(new HttpHost("172.26.17.11", 9202));
//填充host節(jié)點(diǎn)
RestClientBuilder builder = RestClient.builder(httpHosts.toArray(new HttpHost[0]));
builder.setRequestConfigCallback(requestConfigBuilder -> {
requestConfigBuilder.setConnectTimeout(1000);
requestConfigBuilder.setSocketTimeout(1000);
requestConfigBuilder.setConnectionRequestTimeout(1000);
return requestConfigBuilder;
});
//填充用戶名密碼
final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("userName", "password"));
builder.setHttpClientConfigCallback(httpClientBuilder -> {
httpClientBuilder.setMaxConnTotal(30);
httpClientBuilder.setMaxConnPerRoute(30);
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
return httpClientBuilder;
});
restHighLevelClient = new RestHighLevelClient(builder);
}
static {
bulkProcessor=createBulkProcessor();
}
private static BulkProcessor createBulkProcessor() {
BulkProcessor.Listener listener = new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
log.info("1. 【beforeBulk】批次[{}] 攜帶 {} 請求數(shù)量", executionId, request.numberOfActions());
}
@Override
public void afterBulk(long executionId, BulkRequest request,
BulkResponse response) {
if (!response.hasFailures()) {
log.info("2. 【afterBulk-成功】批量 [{}] 完成在 {} ms", executionId, response.getTook().getMillis());
} else {
BulkItemResponse[] items = response.getItems();
for (BulkItemResponse item : items) {
if (item.isFailed()) {
log.info("2. 【afterBulk-失敗】批量 [{}] 出現(xiàn)異常的原因 : {}", executionId, item.getFailureMessage());
break;
}
}
}
}
@Override
public void afterBulk(long executionId, BulkRequest request,
Throwable failure) {
List<DocWriteRequest<?>> requests = request.requests();
List<String> esIds = requests.stream().map(DocWriteRequest::id).collect(Collectors.toList());
log.error("3. 【afterBulk-failure失敗】es執(zhí)行bluk失敗,失敗的esId為：{}", esIds, failure);
}
};
BulkProcessor.Builder builder = BulkProcessor.builder(((bulkRequest, bulkResponseActionListener) -> {
restHighLevelClient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener);
}), listener);
//到達(dá)10000條時刷新
builder.setBulkActions(10000);
//內(nèi)存到達(dá)8M時刷新
builder.setBulkSize(new ByteSizeValue(8L, ByteSizeUnit.MB));
//設(shè)置的刷新間隔10s
builder.setFlushInterval(TimeValue.timeValueSeconds(10));
//設(shè)置允許執(zhí)行的并發(fā)請求數(shù)鸦泳。
builder.setConcurrentRequests(8);
//設(shè)置重試策略
builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1), 3));
return builder.build();
}
}
整個項(xiàng)目可以共用一個BulkProcessor银锻，可以配置多種刷新策略，將數(shù)據(jù)由內(nèi)存刷新到es中做鹰。
3. 增刪改操作
3.1 數(shù)據(jù)準(zhǔn)備
PUT test_demo
PUT test_demo/_mapping
{
"properties":{
"title":{
"type":"text"
},
"tag":{
"type":"keyword"
},
"publishTime":{
"type":"date",
"format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
}
}
}
GET test_demo/_search
{
"query": {
"match_all": {}
}
}
3.2 單條數(shù)據(jù)異步插入
public static void testAsyncSingle() {
IndexRequest indexRequest = new IndexRequest("test_demo");
DemoDto demoDto = new DemoDto(2001L, "印度新冠疫情失控", "世界", new Date());
indexRequest.source(JSON.toJSONString(demoDto), XContentType.JSON);
indexRequest.timeout(TimeValue.timeValueSeconds(1));
indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
//數(shù)據(jù)為存儲而不是更新
indexRequest.create(false);
indexRequest.id(demoDto.getId() + "");
restHighLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
@Override
public void onResponse(IndexResponse indexResponse) {
ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
if (shardInfo.getFailed() > 0) {
for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
log.error("將id為：{}的數(shù)據(jù)存入ES時存在失敗的分片击纬，原因?yàn)椋簕}", indexRequest.id(), failure.getCause());
}
}
}
@Override
public void onFailure(Exception e) {
log.error("{}:存儲es時異常，數(shù)據(jù)信息為", indexRequest.id(), e);
}
});
}
3.3 單條數(shù)據(jù)同步插入
public static void testSingleAdd() throws IOException {
IndexRequest indexRequest = new IndexRequest("test_demo");
DemoDto demoDto = new DemoDto(3001L, "es單數(shù)據(jù)同步插入2", "IT", new Date());
indexRequest.source(JSON.toJSONString(demoDto), XContentType.JSON);
indexRequest.id("3001");
indexRequest.timeout(TimeValue.timeValueSeconds(1));
indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
indexRequest.create(true);
indexRequest.id(demoDto.getId() + "");
restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
}
-
indexRequest.id(demoDto.getId() + ""); —— 填充"_id"字段钾麸。
image.png - indexRequest.create(true);——設(shè)置操作類型
public IndexRequest create(boolean create) {
if (create) {
return opType(OpType.CREATE);
} else {
return opType(OpType.INDEX);
}
}
- OpType.CREATE：當(dāng)存在相同的_id時更振，插入會出現(xiàn)異常；
- OpType.INDEX：當(dāng)存在相同_id時喂走，插入會進(jìn)行覆蓋殃饿；
當(dāng)設(shè)置OpType.CREATE時相同id插入異常看出芋肠，es進(jìn)行了樂觀鎖控制并發(fā)寫沖突乎芳。
Elasticsearch exception [type=version_conflict_engine_exception, reason=[3001]: version conflict, document already exists (current version [3])]
3.4 批量插入
由于設(shè)置了BulkProcessor對象，可以將數(shù)據(jù)設(shè)置到BulkProcessor對象中，根據(jù)策略批量的刷新到Es中奈惑。
/**
* 批量插入
*/
public static void testBatch() {
List<IndexRequest> indexRequests = new ArrayList<>();
ArrayList<DemoDto> demoDtos = new ArrayList<>();
demoDtos.add(new DemoDto(1001L, "中國是中國人的中國", "中國", new Date()));
demoDtos.add(new DemoDto(1002L, "2008年奧運(yùn)會", "體育", new Date()));
demoDtos.forEach(e -> {
IndexRequest request = new IndexRequest("test_demo");
//填充id
request.id(e.getId() + "");
//先不修改id
request.source(JSON.toJSONString(e), XContentType.JSON);
request.opType(DocWriteRequest.OpType.CREATE);
indexRequests.add(request);
});
indexRequests.forEach(bulkProcessor::add);
}
3.5 更新操作
更新操作傳入的doc為map對象吭净，而不是json字符串，否則會拋出異常肴甸。
public static void testSingleUpdate() throws IOException {
UpdateRequest updateRequest = new UpdateRequest("test_demo", "3001");
Map<String, Object> kvs = new HashMap<>();
kvs.put("title", "es單數(shù)據(jù)更新啦寂殉！");
updateRequest.doc(kvs);
updateRequest.timeout(TimeValue.timeValueSeconds(1));
updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
//數(shù)據(jù)為存儲而不是更新
restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
}
3.6 帶條件的更新語句
public static void testSingleUpdateQuery() throws IOException {
UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();
updateByQueryRequest.indices("test_demo");
updateByQueryRequest.setQuery(new TermQueryBuilder("id", 3001));
updateByQueryRequest.setScript(new Script(ScriptType.INLINE,
"painless",
"ctx._source.tag='電腦'", Collections.emptyMap()));
//數(shù)據(jù)為存儲而不是更新
restHighLevelClient.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
}
3.7 批量更新
/**
* 批量更新
*/
private static void testBatchUpdate() {
List<UpdateRequest> updateRequests=new ArrayList<>();
//更新的數(shù)據(jù)
List<DemoDto> params=new ArrayList<>();
params.add(new DemoDto(2001L));
params.add(new DemoDto(3001L));
params.forEach(e->{
//獲取id
UpdateRequest updateRequest = new UpdateRequest();
updateRequest.index("test_demo");
//更新的id
updateRequest.id(e.getId()+"");
//更新的數(shù)據(jù)
Map<String,Object> map=new HashMap<>();
map.put("title","美國社會動蕩");
updateRequest.doc(map);
updateRequests.add(updateRequest);
});
updateRequests.forEach(bulkProcessor::add);
}
3.8 刪除操作
/**
* 單個刪除
*/
private static void testSingleDel() throws IOException {
DeleteRequest deleteRequest=new DeleteRequest();
deleteRequest.index("test_demo");
deleteRequest.id("3001");
restHighLevelClient.delete(deleteRequest,RequestOptions.DEFAULT);
}
3.9 條件刪除
/**
* 單個條件刪除
*/
private static void testSingleDelQuery() throws IOException {
DeleteByQueryRequest deleteByQueryRequest=new DeleteByQueryRequest();
deleteByQueryRequest.indices("test_demo");
deleteByQueryRequest.setQuery(new MatchQueryBuilder("title","國年"));
//分詞式刪除
restHighLevelClient.deleteByQuery(deleteByQueryRequest,RequestOptions.DEFAULT);
}
