個人專題目錄](http://www.reibang.com/p/140e2a59db2c)

1. elasticsearch文檔及索引管理初級

1.1 項目及索引創(chuàng)建

java api 文檔 https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.5.1/java-rest-overview.html

low : 偏向底層。

high：高級封裝。足夠瘫辩。<font color=red>通過API操作與kibana查看操作結(jié)果墓塌。</font>

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.5.1</version>
    <exclusions>
        <exclusion>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>7.5.1</version>
</dependency>

語法：put /index


- title:商品標題
- price:商品價格
- createTime:創(chuàng)建時間
- categoryName:分類名稱。如：家電，手機
- brandName:品牌名稱。如：華為，小米
- spec: 商品規(guī)格瞒大。如： spec:{"屏幕尺寸","5寸"，"內(nèi)存大小","128G"}
- saleNum:銷量
- stock:庫存量

PUT book-index
{
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ik_smart"
            },
            "price": { 
                "type": "double"
            },
            "createTime": {
                "type": "date",
                "format" : "yyyy-MM-dd HH:mm:ss"
            },
            "categoryName": {   
                "type": "keyword"
            },
            "brandName": {  
                "type": "keyword"
            },
    
            "spec": {       
                "type": "object"
            },
            "saleNum": {    
                "type": "integer"
            },
            
            "stock": {  
                "type": "integer"
            }
        }
    }
}

PUT /book-index

@Service
@Log4j2
public class IndexServiceImpl implements IndexService {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Override
    public void createIndex(String index, CreateIndexRequest request, boolean async) throws Exception {
        log.info("source:{}", request.toString());
        //操作索引的客戶端
        IndicesClient indices = restHighLevelClient.indices();
        CreateIndexResponse response = null;
        //要創(chuàng)建索引,首先我們得先判斷索引是不是不存在,如果存在就不創(chuàng)建
        if (!existsIndex(index)) {
            if (async) {
                //異步新增索引
                //監(jiān)聽方法
                ActionListener<CreateIndexResponse> listener = new ActionListener<CreateIndexResponse>() {
                    @Override
                    public void onResponse(CreateIndexResponse createIndexResponse) {
                        log.info("!!!!!!!!創(chuàng)建索引成功" + createIndexResponse.toString());
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("!!!!!!!!創(chuàng)建索引失敗", e);
                    }
                };
                //執(zhí)行創(chuàng)建索引庫
                indices.createAsync(request, RequestOptions.DEFAULT, listener);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                //專門用于 index 相關(guān)的操作
                response = indices.create(request, RequestOptions.DEFAULT);
                //得到響應(yīng)（全部）
                boolean acknowledged = response.isAcknowledged();
                //得到響應(yīng) 指示是否在超時前為索引中的每個分片啟動了所需數(shù)量的碎片副本
                boolean shardsAcknowledged = response.isShardsAcknowledged();
                log.info("創(chuàng)建索引{}的結(jié)果是{}", index, response.isAcknowledged());
            }
        } else {
            log.info("索引已經(jīng)存在{}", index);
        }
    }

    @Override
    public void deleteIndex(String index, boolean async) throws Exception {
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
        IndicesClient indices = restHighLevelClient.indices();
        if (existsIndex(index)) {
            if (async) {
                //異步刪除索引庫
                //監(jiān)聽方法
                ActionListener<AcknowledgedResponse> listener = new ActionListener<AcknowledgedResponse>() {
                    @Override
                    public void onResponse(AcknowledgedResponse deleteIndexResponse) {
                        log.info("!!!!!!!!刪除索引成功 {}", deleteIndexResponse.toString());
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("!!!!!!!!刪除索引失敗", e);
                    }
                };
                //執(zhí)行刪除索引
                indices.deleteAsync(deleteIndexRequest, RequestOptions.DEFAULT, listener);

                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                //如果index 存在就刪除
                //創(chuàng)建用于刪除索引的請求
                AcknowledgedResponse response = indices.delete(deleteIndexRequest, RequestOptions.DEFAULT);
                log.info("刪除索引{}的結(jié)果是{}", index, response.isAcknowledged());
            }

        } else {
            log.info("索引不存在{},無法刪除", index);
        }
    }

    @Override
    public boolean existsIndex(String index) throws Exception {
        //設(shè)置要查詢的索引
        GetIndexRequest getIndexRequest = new GetIndexRequest(index);
        IndicesClient indices = restHighLevelClient.indices();
        //從主節(jié)點返回本地信息或檢索狀態(tài)
        getIndexRequest.local(false);
        //以適合人類的格式返回結(jié)果
        getIndexRequest.humanReadable(true);
        //是否返回每個索引的所有默認設(shè)置
        getIndexRequest.includeDefaults(false);
        boolean exists = indices.exists(getIndexRequest, RequestOptions.DEFAULT);
        log.info("索引{}存在的狀態(tài)是{}", index, exists);
        return exists;
    }


    @Override
    public void openIndex(String indexName) throws IOException {
        OpenIndexRequest request = new OpenIndexRequest(indexName);
        IndicesClient indices = restHighLevelClient.indices();
        OpenIndexResponse openIndexResponse = indices.open(request, RequestOptions.DEFAULT);
        boolean acknowledged = openIndexResponse.isAcknowledged();
        log.info("!!!!!!!!!" + acknowledged);
    }

    @Override
    public void closeIndex(String indexName) throws IOException {
        CloseIndexRequest request = new CloseIndexRequest(indexName);
        CloseIndexResponse closeIndexResponse = restHighLevelClient.indices().close(request, RequestOptions.DEFAULT);
        boolean acknowledged = closeIndexResponse.isAcknowledged();
        log.info("!!!!!!!!!" + acknowledged);
    }
}

1.2 索引操作示例

@SpringBootTest(classes = SearchServiceApplication.class)
@WebAppConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class IndexServiceTest {

    @Autowired
    private IndexService indexService;

    @Test
    public void testCreateIndex() throws Exception {
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(Constants.INDEX_NAME);
        //我們創(chuàng)建index 和 type 的 時候需要指定分配和 mapping
        buildingSetting(createIndexRequest);
        buildingMapping(createIndexRequest);

        //設(shè)置別名
        //createIndexRequest.alias(new Alias("alias_index_name"));

        // 額外參數(shù)
        //設(shè)置超時時間
        createIndexRequest.setTimeout(TimeValue.timeValueMinutes(2));
        //設(shè)置主節(jié)點超時時間
        createIndexRequest.setMasterTimeout(TimeValue.timeValueMinutes(1));
        //在創(chuàng)建索引API返回響應(yīng)之前等待的活動分片副本的數(shù)量搪桂，以int形式表示
        createIndexRequest.waitForActiveShards(ActiveShardCount.from(2));
        createIndexRequest.waitForActiveShards(ActiveShardCount.DEFAULT);

        indexService.createIndex(Constants.INDEX_NAME, createIndexRequest, false);
    }

    /**
     * - title:商品標題
     * - price:商品價格
     * - createTime:創(chuàng)建時間
     * - categoryName:分類名稱透敌。如：家電，手機
     * - brandName:品牌名稱。如：華為酗电，小米
     * - spec: 商品規(guī)格淌山。如： spec:{"屏幕尺寸","5寸"，"內(nèi)存大小","128G"}
     * - saleNum:銷量
     * - stock:庫存量
     */
    private void buildingMapping(CreateIndexRequest createIndexRequest) throws IOException {
        XContentBuilder xContentBuilder = JsonXContent.contentBuilder()
                .startObject()
                .startObject("properties")

                .startObject("title")
                .field("type", "text")
                .field("analyzer", "ik_smart")
                .endObject()

                .startObject("price")
                .field("type", "double")
                .endObject()

                .startObject("createTime")
                .field("type", "date")
                .field("format", "yyyy-MM-dd HH:mm:ss")
                .endObject()

                .startObject("categoryName")
                .field("type", "keyword")
                .endObject()

                .startObject("brandName")
                .field("type", "keyword")
                .endObject()

                .startObject("spec")
                .field("type", "object")
                .endObject()

                .startObject("saleNum")
                .field("type", "integer")
                .endObject()

                .startObject("stock")
                .field("type", "integer")
                .endObject()

                .endObject()
                .endObject();
        createIndexRequest.mapping(xContentBuilder);

    }

    /**
     * 設(shè)置 index 的分片規(guī)則
     *
     * @param createIndexRequest
     */
    private void buildingSetting(CreateIndexRequest createIndexRequest) {
        createIndexRequest.settings(Settings.builder()
                // 設(shè)置主分片為 3
                .put("number_of_shards", 3)
                //設(shè)置從分片為 2
                .put("number_of_replicas", 2));
    }

    @Test
    public void testDeleteIndex() throws Exception {
        indexService.deleteIndex(Constants.INDEX_NAME, false);
    }

    @Test
    public void testExistsIndex() throws Exception {
        indexService.existsIndex(Constants.INDEX_NAME);
    }

    @Test
    public void testOpenIndex() throws IOException {
        indexService.openIndex(Constants.INDEX_NAME);
    }

    @Test
    public void testCloseIndex() throws IOException {
        indexService.closeIndex(Constants.INDEX_NAME);
    }
}

1.3 查詢文檔

語法：GET /index/type/id

查看:GET /book-index/ 就可看到j(luò)son形式的文檔顾瞻。方便程序解析泼疑。

_mget批量查詢

批量查詢可以提高查詢效率。推薦使用（相對于單數(shù)據(jù)查詢來說）荷荤。

@Override
public void get(String index, String id, boolean async) throws Exception {
    //構(gòu)建請求
    GetRequest getRequest = new GetRequest(index, id);

    //可選參數(shù)
    //為特定字段配置_source_include
    String[] includes = new String[]{"id", "price"};
    String[] excludes = Strings.EMPTY_ARRAY;
    FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
    getRequest.fetchSourceContext(fetchSourceContext);

    //設(shè)置路由
    //getRequest.routing("id");

    if (async) {
        // 執(zhí)行 查詢 同步查詢
        GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
        // 獲取結(jié)果
        if (getResponse.isExists()) {
            long version = getResponse.getVersion();
            //檢索文檔(String形式)
            String sourceAsString = getResponse.getSourceAsString();
            log.info(sourceAsString);
            //以字節(jié)接受
            byte[] sourceAsBytes = getResponse.getSourceAsBytes();
            Map<String, Object> sourceAsMap = getResponse.getSourceAsMap();
            log.info(sourceAsMap);
        }
    } else {
        //異步查詢
        ActionListener<GetResponse> listener = new ActionListener<GetResponse>() {
            //查詢成功時的立馬執(zhí)行的方法
            @Override
            public void onResponse(GetResponse getResponse) {
                long version = getResponse.getVersion();
                //檢索文檔(String形式)
                String sourceAsString = getResponse.getSourceAsString();
                log.info(sourceAsString);
            }

            //查詢失敗時的立馬執(zhí)行的方法
            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
            }
        };
        //執(zhí)行異步請求
        restHighLevelClient.getAsync(getRequest, RequestOptions.DEFAULT, listener);
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

@Override
public void mGet(List<BulkBean> beans) throws Exception {
    MultiGetRequest multiGetRequest = new MultiGetRequest();
    for (BulkBean bean : beans) {
        multiGetRequest.add(bean.getIndex(), bean.getId());
    }
    MultiGetResponse multiGetResponse = restHighLevelClient.mget(multiGetRequest, RequestOptions.DEFAULT);
    //獲取響應(yīng)
    MultiGetItemResponse[] responses = multiGetResponse.getResponses();
    for (MultiGetItemResponse response : responses) {
        //將數(shù)據(jù)以 map 格式展示
        Map<String, Object> map = response.getResponse().getSource();
        //將數(shù)據(jù)以 json 格式展示
        String source = response.getResponse().getSourceAsString();
        log.info("每一條數(shù)據(jù)是{}" + source);
    }
    log.info(multiGetResponse);
}

1.3 新建文檔

語法：PUT /index/_doc/id

為防止覆蓋原有數(shù)據(jù)退渗，我們在新增時，設(shè)置為強制創(chuàng)建蕴纳，不會覆蓋原有文檔会油。

語法：PUT /index/ _doc/id/_create

使用強制新增語法時，如果Document的id在Elasticsearch中已存在古毛，則會報錯翻翩。（version conflict, document already exists）

# 此操作為Elasticsearch自動生成id的新增Document方式。
POST book-index/_doc/1
{
  "title":"小米手機",
  "price":1000,
  "createTime":"2019-12-01",
  "categoryName":"手機",
  "brandName":"小米",
  "saleNum":3000,
  "stock":10000,
  "spec":{
    "網(wǎng)絡(luò)制式":"移動4G",
    "屏幕尺寸":"4.5"
  }
}
#指定id創(chuàng)建文檔,需要 PUT 請求
PUT /book-index/1
{

}

public void addDoc(String index, String json, String docId) throws IOException {
    IndexRequest indexRequest = new IndexRequest(index);
    //設(shè)置我們要傳遞的數(shù)據(jù)
    indexRequest.source(json, XContentType.JSON);
    if (docId != null) {
        indexRequest.id(docId);
    }
    //可選參數(shù)
    //設(shè)置超時時間
    indexRequest.timeout(TimeValue.timeValueSeconds(1));
    indexRequest.timeout("1s");

    //自己維護版本號
    //indexRequest.version(2);
    //indexRequest.versionType(VersionType.EXTERNAL);

    IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
    log.info("添加數(shù)據(jù)indexResponse {}" + objectMapper.writeValueAsString(indexResponse));

    //構(gòu)建方法2
//        XContentBuilder builder = XContentFactory.jsonBuilder();
//        builder.startObject();
//        {
//            builder.field("user", "tomas");
//            builder.timeField("postDate", new Date());
//            builder.field("message", "trying out es2");
//        }
//        builder.endObject();
//        indexRequest.source(builder);
//        //異步
//        ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
//            @Override
//            public void onResponse(IndexResponse indexResponse) {
//
//            }
//
//            @Override
//            public void onFailure(Exception e) {
//
//            }
//        };
//        client.indexAsync(indexRequest, RequestOptions.DEFAULT, listener);
//        try {
//            Thread.sleep(5000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

    String indexName = indexResponse.getIndex();
    String id = indexResponse.getId();
    //獲取插入的類型
    if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
        DocWriteResponse.Result result = indexResponse.getResult();
        log.info("CREATED:" + result);
    } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
        DocWriteResponse.Result result = indexResponse.getResult();
        log.info("UPDATED:" + result);
    }

    ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
    if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
        log.info("處理成功的分片數(shù)少于總分片稻薇！");
    }
    if (shardInfo.getFailed() > 0) {
        for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
            //處理潛在的失敗原因
            String reason = failure.reason();
            log.info(reason);
        }
    }
}

1.4 更新文檔

PUT /index/type/1 替換操作是整體覆蓋嫂冻，要帶上所有信息。

執(zhí)行兩次塞椎，返回結(jié)果中版本號（_version）在不斷上升桨仿。此過程為全量替換。

實質(zhì)：舊文檔的內(nèi)容不會立即刪除案狠，只是標記為deleted服傍。適當?shù)臅r機，集群會將這些文檔刪除骂铁。

局部替換 partial update

POST方式更新單個內(nèi)容

語法：POST /{index}/type /{id}/_update

或者POST /{index}/_update/{id}

partial update局部替換則只修改變動字段吹零。

內(nèi)部與全量替換是一樣的，舊文檔標記為刪除拉庵，新建一個文檔灿椅。

優(yōu)點：

• 大大減少網(wǎng)絡(luò)傳輸次數(shù)和流量，提升性能
• 減少并發(fā)沖突發(fā)生的概率名段。

POST /book-index/1/_update
{
  "doc":{
  "ipAddr":"10.126.2.9"
  }
}

public UpdateResponse update(String index, String type, Map<String, Object> values, String id) throws IOException {
    //創(chuàng)建更新請求,并指定 index,type 和 id  局部更新部分數(shù)據(jù)
    UpdateRequest updateRequest = new UpdateRequest(index, type, id).doc(values);

    //可選參數(shù)
    //超時時間
    updateRequest.timeout("1s");

    //重試次數(shù)
    updateRequest.retryOnConflict(3);
    //設(shè)置在繼續(xù)更新之前阱扬，必須激活的分片數(shù)
    updateRequest.waitForActiveShards(2);
    //所有分片都是active狀態(tài)，才更新
    updateRequest.waitForActiveShards(ActiveShardCount.ALL);
    UpdateResponse updateResponse = client.update(updateRequest, RequestOptions.DEFAULT);
    updateResponse.getId();
    updateResponse.getIndex();
    //判斷結(jié)果
    if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
        DocWriteResponse.Result result = updateResponse.getResult();
        System.out.println("CREATED:" + result);
    } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
        DocWriteResponse.Result result = updateResponse.getResult();
        System.out.println("UPDATED:" + result);
    } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
        DocWriteResponse.Result result = updateResponse.getResult();
        System.out.println("DELETED:" + result);
    } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
        //沒有操作
        DocWriteResponse.Result result = updateResponse.getResult();
        System.out.println("NOOP:" + result);
    }
    return updateResponse;
}

1.5 刪除文檔

Elasticsearch中執(zhí)行刪除操作時伸辟，Elasticsearch先標記Document為deleted狀態(tài)，而不是直接物理刪除馍刮。當Elasticsearch存儲空間不足或工作空閑時信夫，才會執(zhí)行物理刪除操作。標記為deleted狀態(tài)的數(shù)據(jù)不會被查詢搜索到。

DELETE /book-index/1

public void deleteDocById(String index, String id) throws Exception {
    DeleteRequest deleteRequest = new DeleteRequest(index, id);

    DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
    deleteResponse.getId();
    deleteResponse.getIndex();
    DocWriteResponse.Result result = deleteResponse.getResult();
    log.info(result);
    log.info("刪除狀態(tài)是{}", deleteResponse.status().getStatus());
}

1.6 批量操作bulk

注意：bulk語法中要求一個完整的json串不能有換行静稻。不同的json串必須使用換行分隔警没。多個操作中，如果有錯誤情況振湾，不會影響到其他的操作杀迹，只會在批量操作返回結(jié)果中標記失敗。bulk語法批量操作時押搪，bulk request會一次性加載到內(nèi)存中树酪，如果請求數(shù)據(jù)量太大，性能反而下降（內(nèi)存壓力過高）大州，需要反復嘗試一個最佳的bulk request size续语。一般從1000~5000條數(shù)據(jù)開始嘗試，逐漸增加厦画。如果查看bulk request size的話疮茄，一般是5~15MB之間為好。

bulk語法要求json格式是為了對內(nèi)存的方便管理根暑，和盡可能降低內(nèi)存的壓力力试。如果json格式?jīng)]有特殊的限制，Elasticsearch在解釋bulk請求時排嫌，需要對任意格式的json進行解釋處理懂版，需要對bulk請求數(shù)據(jù)做json對象會json array對象的轉(zhuǎn)化，那么內(nèi)存的占用量至少翻倍躏率，當請求量過大的時候躯畴，對內(nèi)存的壓力會直線上升，且需要jvm gc進程對垃圾數(shù)據(jù)做頻繁回收薇芝，影響Elasticsearch效率蓬抄。

生產(chǎn)環(huán)境中，bulk api常用夯到。都是使用java代碼實現(xiàn)循環(huán)操作嚷缭。一般一次bulk請求，執(zhí)行一種操作耍贾。如：批量新增10000條數(shù)據(jù)等阅爽。

POST /_bulk
{"action": {"metadata"}}
{"data"}


#bulk 批量添加,批量的時候第一行為id 列,第二行為數(shù)據(jù)列,中間不能出現(xiàn)換行
POST /book-index/_doc/_bulk
{"index":{"_id":1}}
{"corpName":"途虎養(yǎng)車",...}
{"index":{"_id":2}}
{"corpName":"盒馬鮮生"...}


#可以刪除不同 index 下的數(shù)據(jù),下面案例不演示了,注意會在倒數(shù)第二行報錯,因為第一條就是刪除的它,即便在地址中指定了庫,可以去刪除其他index 的數(shù)據(jù),在參數(shù)中不指定 index 的情況下就是按照地址中的來,指定了 index 的情況下就是按照具體指定的來
POST /lib2/books/_bulk
{"delete":{"_index":"lib2","_type":"books","_id":4}}
{"create":{"_index":"tt","_type":"ttt","_id":"100"}}
{"name":"lisi"}
{"index":{"_index":"tt","_type":"ttt"}}
{"name":"zhaosi"}
{"update":{"_index":"lib2","_type":"books","_id":"4"}}
{"doc":{"price":58}}


#可以指定不同的 index 和 type
GET /_mget
{
"docs":[
   {
       "_index": "book-index", #索引
       "_type": "_doc", #數(shù)據(jù)類型
       "_id": 1 #要查詢的主鍵
   },
     {
       "_index": "book-index", 
       "_type": "_doc",
       "_id": 2
   }
 ]
}

action:(行為)

create：文檔不存在時創(chuàng)建

update:更新文檔

index:創(chuàng)建新文檔或替換已有文檔

delete:刪除一個文檔

metadata：_index,_type,_id

create 和index的區(qū)別

如果數(shù)據(jù)存在，使用create操作失敗荐开，會提示文檔已經(jīng)存在付翁，使用index則可以成功執(zhí)行。

public void bulkOption(List<BulkBeanWithOption> beanWithOptionList) throws IOException {
    BulkRequest bulkRequest = new BulkRequest();

    for (BulkBeanWithOption bean : beanWithOptionList) {
        switch (bean.getBulkOption()) {
            //根據(jù)我們的操作類型來決定做什么
            case INDEX:
                IndexRequest indexRequest = new IndexRequest(bean.getIndex());
                indexRequest.id(bean.getId());
                indexRequest.source(bean.getJson(), XContentType.JSON);
                bulkRequest.add(indexRequest);
                break;
            case CREATE:
                break;
            case DELETE:
                DeleteRequest deleteRequest = new DeleteRequest(bean.getIndex());
                deleteRequest.id(bean.getId());
                bulkRequest.add(deleteRequest);
                break;
            case UPDATE:
                log.info("update");
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + bean.getBulkOption());
        }
    }
    BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);

    for (BulkItemResponse itemResponse : bulkResponse) {
        DocWriteResponse itemResponseResponse = itemResponse.getResponse();

        switch (itemResponse.getOpType()) {
            case INDEX:
            case CREATE:
                IndexResponse indexResponse = (IndexResponse) itemResponseResponse;
                indexResponse.getId();
                log.info(indexResponse.getResult());
                break;
            case UPDATE:
                UpdateResponse updateResponse = (UpdateResponse) itemResponseResponse;
                updateResponse.getIndex();
                log.info(updateResponse.getResult());
                break;
            case DELETE:
                DeleteResponse deleteResponse = (DeleteResponse) itemResponseResponse;
                log.info(deleteResponse.getResult());
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + itemResponse.getOpType());
        }
    }
}

1.7 索引refresh

一個理想的搜索解決方案中晃听，新索引的數(shù)據(jù)應(yīng)該能立即搜索到百侧。ElasticSearch給人的第一印象仿佛就是如此工作的砰识，即使是在多服務(wù)器環(huán)境下，然而事實并非如此（至少不是任何場景都能保證新索引的數(shù)據(jù)能被實時檢索到）佣渴。

elasticsearch是基于lucene的辫狼，lucene是可以做到實時的，就是創(chuàng)建索引之后辛润，立即能查詢到膨处。但是這樣，要么是犧牲索引的效率砂竖，每次都索引之后都刷新真椿，要么就是犧牲查詢的效率每次查詢之前都進行刷新。

無論哪一種晦溪，都會讓你的性能下降10倍以上瀑粥，所以只能采取一種折中的方案，每隔n秒自動刷新三圆，這樣你創(chuàng)建索引之后狞换，最多在ns之內(nèi)肯定能查到。這就是所謂的準實時(near real-time)查詢舟肉。

elasticsearch默認刷新時間是1s修噪。

刷新索引方法：

public void refreshIndex(String indexName) throws IOException {
    RefreshRequest refreshRequest = new RefreshRequest(indexName);
    IndicesClient indices = restHighLevelClient.indices();
    RefreshResponse refresh = indices.refresh(refreshRequest, RequestOptions.DEFAULT);
    log.info(refresh.toString());
}