elasticsearch 源码 创建索引_es索引针源码

elasticsearch 源码 创建索引

创建索引过程，当我们客户端提交一个创建索引请求时，之前提到了es的transport模块，在处理请求时，会将请求分发到对应的TransportRequestHandler，而创建索引的入口就是TransportHandler对象，这个对象对应的类是TransportCreateIndexAction的内部类，而TransportCreateIndexAction的这个内部类继承自父类TransportMasterNodeOperationAction。

//TransportMasterNodeOperationAction.java
private class TransportHandler extends BaseTransportRequestHandler<Request> {

    @Override
    public Request newInstance() {
        return newRequest();
    }

    @Override
    public String executor() {
        return ThreadPool.Names.SAME;
    }  // 使用的是same，请求不是交由线程池执行

    @Override
    public void messageReceived(final Request request, final TransportChannel channel) throws Exception {
        // we just send back a response, no need to fork a listener
        request.listenerThreaded(false);
        execute(request, new ActionListener<Response>() {
            @Override
            public void onResponse(Response response) {
                try {
                    channel.sendResponse(response);
                } catch (Throwable e) {
                    onFailure(e);
                }
            }

            @Override
            public void onFailure(Throwable e) {
                try {
                    channel.sendResponse(e);
                } catch (Exception e1) {
                    logger.warn("Failed to send response", e1);
                }
            }
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

服务端在接收到创建索引请求时，会调用TransportHandler的messageReceived方法，进而调用execute方法。可以看到ActionListener 封装了response，让调用者可以执行调用OnResponse方法来响应，或者调用OnFailure来响应失败。而后

//TransportMasterNodeOperationAction.java
public void execute(Request request, ActionListener<Response> listener) {
    request.listenerThreaded(true);
    super.execute(request, listener);
}

//TransportAction.java
public void execute(Request request, ActionListener<Response> listener) {
    if (request.listenerThreaded()) {
        //对ActionListener 进行封装ThreadedActio
1
2
3
4
5
6
7
8
9