import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public HttpInvoker(Settings settings, ThreadPool threadPool, Headers headers, URL url) {
    super(settings, threadPool, headers);
    this.contexts = new HashMap<>();
    this.bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(
            Executors.newCachedThreadPool(),
            Executors.newCachedThreadPool()));
    bootstrap.setPipelineFactory(new HttpInvoker.HttpClientPipelineFactory());
    bootstrap.setOption("tcpNoDelay", true);

    registerAction(BulkAction.INSTANCE, HttpBulkAction.class);
    registerAction(CreateIndexAction.INSTANCE, HttpCreateIndexAction.class);
    registerAction(RefreshAction.INSTANCE, HttpRefreshIndexAction.class);
    registerAction(ClusterUpdateSettingsAction.INSTANCE, HttpClusterUpdateSettingsAction.class);
    registerAction(UpdateSettingsAction.INSTANCE, HttpUpdateSettingsAction.class);
    registerAction(SearchAction.INSTANCE, HttpSearchAction.class);

    this.url = url;
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public HttpElasticsearchClient build() {
    if (url == null && host != null && port != null) {
        try {
            url = new URL("http://" + host + ":" + port);
        } catch (MalformedURLException e) {
            throw new IllegalArgumentException("malformed url: " + host + ":" + port);
        }
    }
    if (url == null) {
        throw new IllegalArgumentException("no base URL given");
    }
    ThreadPool threadpool = new ThreadPool("http_client_pool");
    client = new HttpElasticsearchClient(settings, threadpool, Headers.EMPTY, url);

    client.registerAction(BulkAction.INSTANCE, HttpBulkAction.class);
    client.registerAction(CreateIndexAction.INSTANCE, HttpCreateIndexAction.class);
    client.registerAction(RefreshAction.INSTANCE, HttpRefreshIndexAction.class);
    client.registerAction(ClusterUpdateSettingsAction.INSTANCE, HttpClusterUpdateSettingsAction.class);
    client.registerAction(UpdateSettingsAction.INSTANCE, HttpUpdateSettingsAction.class);
    client.registerAction(SearchAction.INSTANCE, HttpSearchAction.class);

    return client;
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void testIndex() {
    String[] indexShardActions = new String[]{BulkAction.NAME + "[s][p]", BulkAction.NAME + "[s][r]"};
    interceptTransportActions(indexShardActions);

    IndexRequest indexRequest = new IndexRequest(randomIndexOrAlias(), "type", "id")
        .source(Requests.INDEX_CONTENT_TYPE, "field", "value");
    internalCluster().coordOnlyNodeClient().index(indexRequest).actionGet();

    clearInterceptedActions();
    assertSameIndices(indexRequest, indexShardActions);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void testDelete() {
    String[] deleteShardActions = new String[]{BulkAction.NAME + "[s][p]", BulkAction.NAME + "[s][r]"};
    interceptTransportActions(deleteShardActions);

    DeleteRequest deleteRequest = new DeleteRequest(randomIndexOrAlias(), "type", "id");
    internalCluster().coordOnlyNodeClient().delete(deleteRequest).actionGet();

    clearInterceptedActions();
    assertSameIndices(deleteRequest, deleteShardActions);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void testUpdate() {
    //update action goes to the primary, index op gets executed locally, then replicated
    String[] updateShardActions = new String[]{UpdateAction.NAME + "[s]", BulkAction.NAME + "[s][p]", BulkAction.NAME + "[s][r]"};
    interceptTransportActions(updateShardActions);

    String indexOrAlias = randomIndexOrAlias();
    client().prepareIndex(indexOrAlias, "type", "id").setSource("field", "value").get();
    UpdateRequest updateRequest = new UpdateRequest(indexOrAlias, "type", "id").doc(Requests.INDEX_CONTENT_TYPE, "field1", "value1");
    UpdateResponse updateResponse = internalCluster().coordOnlyNodeClient().update(updateRequest).actionGet();
    assertEquals(DocWriteResponse.Result.UPDATED, updateResponse.getResult());

    clearInterceptedActions();
    assertSameIndices(updateRequest, updateShardActions);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void testUpdateUpsert() {
    //update action goes to the primary, index op gets executed locally, then replicated
    String[] updateShardActions = new String[]{UpdateAction.NAME + "[s]", BulkAction.NAME + "[s][p]", BulkAction.NAME + "[s][r]"};
    interceptTransportActions(updateShardActions);

    String indexOrAlias = randomIndexOrAlias();
    UpdateRequest updateRequest = new UpdateRequest(indexOrAlias, "type", "id").upsert(Requests.INDEX_CONTENT_TYPE, "field", "value")
        .doc(Requests.INDEX_CONTENT_TYPE, "field1", "value1");
    UpdateResponse updateResponse = internalCluster().coordOnlyNodeClient().update(updateRequest).actionGet();
    assertEquals(DocWriteResponse.Result.CREATED, updateResponse.getResult());

    clearInterceptedActions();
    assertSameIndices(updateRequest, updateShardActions);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void testUpdateDelete() {
    //update action goes to the primary, delete op gets executed locally, then replicated
    String[] updateShardActions = new String[]{UpdateAction.NAME + "[s]", BulkAction.NAME + "[s][p]", BulkAction.NAME + "[s][r]"};
    interceptTransportActions(updateShardActions);

    String indexOrAlias = randomIndexOrAlias();
    client().prepareIndex(indexOrAlias, "type", "id").setSource("field", "value").get();
    UpdateRequest updateRequest = new UpdateRequest(indexOrAlias, "type", "id")
            .script(new Script(ScriptType.INLINE, CustomScriptPlugin.NAME, "ctx.op='delete'", Collections.emptyMap()));
    UpdateResponse updateResponse = internalCluster().coordOnlyNodeClient().update(updateRequest).actionGet();
    assertEquals(DocWriteResponse.Result.DELETED, updateResponse.getResult());

    clearInterceptedActions();
    assertSameIndices(updateRequest, updateShardActions);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void execute(BulkRequest bulkRequest, long executionId) {
    boolean afterCalled = false;
    try {
        listener.beforeBulk(executionId, bulkRequest);
        BulkResponse bulkResponse = client.execute(BulkAction.INSTANCE, bulkRequest).actionGet();
        afterCalled = true;
        listener.afterBulk(executionId, bulkRequest, bulkResponse);
    } catch (Throwable t) {
        if (!afterCalled) {
            listener.afterBulk(executionId, bulkRequest, t);
        }
    }
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
@Override
public ActionFuture<BulkResponse> bulk(final BulkRequest request) {
    return execute(BulkAction.INSTANCE, request);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
@Override
public BulkRequestBuilder prepareBulk() {
    return new BulkRequestBuilder(this, BulkAction.INSTANCE);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public void testTransportBulkTasks() {
    registerTaskManageListeners(BulkAction.NAME);  // main task
    registerTaskManageListeners(BulkAction.NAME + "[s]");  // shard task
    registerTaskManageListeners(BulkAction.NAME + "[s][p]");  // shard task on primary
    registerTaskManageListeners(BulkAction.NAME + "[s][r]");  // shard task on replica
    createIndex("test");
    ensureGreen("test"); // Make sure all shards are allocated to catch replication tasks
    client().prepareBulk().add(client().prepareIndex("test", "doc", "test_id")
        .setSource("{\"foo\": \"bar\"}", XContentType.JSON)).get();

    // the bulk operation should produce one main task
    List<TaskInfo> topTask = findEvents(BulkAction.NAME, Tuple::v1);
    assertEquals(1, topTask.size());
    assertEquals("requests[1], indices[test]", topTask.get(0).getDescription());

    // we should also get 1 or 2 [s] operation with main operation as a parent
    // in case the primary is located on the coordinating node we will have 1 operation, otherwise - 2
    List<TaskInfo> shardTasks = findEvents(BulkAction.NAME + "[s]", Tuple::v1);
    assertThat(shardTasks.size(), allOf(lessThanOrEqualTo(2), greaterThanOrEqualTo(1)));

    // Select the effective shard task
    TaskInfo shardTask;
    if (shardTasks.size() == 1) {
        // we have only one task - it's going to be the parent task for all [s][p] and [s][r] tasks
        shardTask = shardTasks.get(0);
        // and it should have the main task as a parent
        assertParentTask(shardTask, findEvents(BulkAction.NAME, Tuple::v1).get(0));
        assertEquals("requests[1], index[test]", shardTask.getDescription());
    } else {
        if (shardTasks.get(0).getParentTaskId().equals(shardTasks.get(1).getTaskId())) {
            // task 1 is the parent of task 0, that means that task 0 will control [s][p] and [s][r] tasks
             shardTask = shardTasks.get(0);
            // in turn the parent of the task 1 should be the main task
            assertParentTask(shardTasks.get(1), findEvents(BulkAction.NAME, Tuple::v1).get(0));
            assertEquals("requests[1], index[test]", shardTask.getDescription());
        } else {
            // otherwise task 1 will control [s][p] and [s][r] tasks
            shardTask = shardTasks.get(1);
            // in turn the parent of the task 0 should be the main task
            assertParentTask(shardTasks.get(0), findEvents(BulkAction.NAME, Tuple::v1).get(0));
            assertEquals("requests[1], index[test]", shardTask.getDescription());
        }
    }

    // we should also get one [s][p] operation with shard operation as a parent
    assertEquals(1, numberOfEvents(BulkAction.NAME + "[s][p]", Tuple::v1));
    assertParentTask(findEvents(BulkAction.NAME + "[s][p]", Tuple::v1), shardTask);

    // we should get as many [s][r] operations as we have replica shards
    // they all should have the same shard task as a parent
    assertEquals(getNumShards("test").numReplicas, numberOfEvents(BulkAction.NAME + "[s][r]", Tuple::v1));
    assertParentTask(findEvents(BulkAction.NAME + "[s][r]", Tuple::v1), shardTask);
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
public BulkAction getAction() {
    return BulkAction.INSTANCE;
}
 

import org.elasticsearch.action.bulk.BulkAction; //导入依赖的package包/类
@Test
public void testSearch() throws Exception {
    Client client = client("1");
    long t0 = System.currentTimeMillis();
    BulkRequestBuilder builder = new BulkRequestBuilder(client, BulkAction.INSTANCE);
    for (int i = 0; i < 1000; i++) {
        builder.add(indexRequest()
                .index("pages").type("row")
                .source(jsonBuilder()
                        .startObject()
                        .field("user1", "kimchy")
                        .field("user2", "kimchy")
                        .field("user3", "kimchy")
                        .field("user4", "kimchy")
                        .field("user5", "kimchy")
                        .field("user6", "kimchy")
                        .field("user7", "kimchy")
                        .field("user8", "kimchy")
                        .field("user9", "kimchy")
                        .field("rowcount", i)
                        .field("rs", 1234)));
    }
    client.bulk(builder.request()).actionGet();

    client.admin().indices().refresh(refreshRequest()).actionGet();

    long t1 = System.currentTimeMillis();
    logger.info("t1-t0 = {}", t1-t0);

    for (int i = 0; i < 100; i++) {
        t1 = System.currentTimeMillis();
        QueryBuilder queryStringBuilder =
                QueryBuilders.queryStringQuery("rs:" + 1234);
        SearchRequestBuilder requestBuilder = client.prepareSearch()
                .setIndices("pages")
                .setTypes("row")
                .setQuery(queryStringBuilder)
                .addSort("rowcount", SortOrder.DESC)
                .setFrom(i*10).setSize(10);
        SearchResponse response = requestBuilder.execute().actionGet();
        long t2 = System.currentTimeMillis();
        logger.info("t2-t1 = {}", t2-t1);
    }
}