1 请求URL 与对应Action

如果我们发送如下的http请求，elasticsearch是如何匹配对应Action的呢？

curl -XGET 'http://elasticsearch_ip:port/xxx/url/path' -d '{
    "请求的content"
}'

在org.elasticsearch.rest.action包下面，存放了处理各种请求的rest action类
在每个action的构造方法中，会将当前action对象注册到指定的url上

public class RestXxxAction extends BaseRestHandler {

    @Inject
    public RestXxxAction(Settings settings, RestController controller, Client client) {
        super(settings, controller, client);
        controller.registerHandler(GET, "/xxx/url/path", this);
        controller.registerHandler(POST, "/xxx/url/path", this);
    }
}

在registerHandler()方法中，会将url path和对应handler添加到http method对象上

public class RestController extends AbstractLifecycleComponent<RestController> {
    public void registerHandler(RestRequest.Method method, String path, RestHandler handler) {
        switch (method) {
            case GET:
                getHandlers.insert(path, handler);
                break;
            case DELETE:
                deleteHandlers.insert(path, handler);
                break;
            case POST:
                postHandlers.insert(path, handler);
                break;
            case PUT:
                putHandlers.insert(path, handler);
                break;
            case OPTIONS:
                optionsHandlers.insert(path, handler);
                break;
            case HEAD:
                headHandlers.insert(path, handler);
                break;
            default:
                throw new ElasticsearchIllegalArgumentException("Can't handle [" + method + "] for path [" + path + "]");
        }
    }
}

elasticsearch使用HttpRequestHandler接收http请求消息，最终会在executeHandler()方法中调用getHandler()方法获取path对应的handler

public class RestController extends AbstractLifecycleComponent<RestController> {
    void executeHandler(RestRequest request, RestChannel channel) throws Exception {
        final RestHandler handler = getHandler(request);
        if (handler != null) {
            handler.handleRequest(request, channel);
        } else {
            if (request.method() == RestRequest.Method.OPTIONS) {
                channel.sendResponse(new BytesRestResponse(OK));
            } else {
                channel.sendResponse(new BytesRestResponse(BAD_REQUEST, 
  "No handler found for uri [" + request.uri() + "] and method [" + request.method() + "]"));
            }
        }
    }
}

在getHandler()方法中，主要是根据method和path获取在RestXxxAction中添加的handler

public class RestController extends AbstractLifecycleComponent<RestController> {
    private RestHandler getHandler(RestRequest request) {
        String path = getPath(request);
        RestRequest.Method method = request.method();
        if (method == RestRequest.Method.GET) {
            return getHandlers.retrieve(path, request.params());
        } else if (method == RestRequest.Method.POST) {
            return postHandlers.retrieve(path, request.params());
        } else if (method == RestRequest.Method.PUT) {
            return putHandlers.retrieve(path, request.params());
        } else if (method == RestRequest.Method.DELETE) {
            return deleteHandlers.retrieve(path, request.params());
        } else if (method == RestRequest.Method.HEAD) {
            return headHandlers.retrieve(path, request.params());
        } else if (method == RestRequest.Method.OPTIONS) {
            return optionsHandlers.retrieve(path, request.params());
        } else {
            return null;
        }
    }
}

elasticsearch在path对应获取Handler后，就执行该handler的handleRequest()方法，作为rest逻辑入口

2 Action与对应TransportAction

在ActionModule中，会将Action和对应的TransportAction注册到actions对象中

public class ActionModule extends AbstractModule {
    @Override
    protected void configure() {
        registerAction(XxxAction.INSTANCE, TransportXxxAction.class);
    }

    public <Request extends ActionRequest, Response extends ActionResponse> void registerAction(
                        GenericAction<Request, 
                        Response> action, Class<? extends TransportAction<Request, Response>> transportAction, 
                        Class... supportTransportActions) {
        actions.put(action.name(), new ActionEntry<>(action, transportAction, supportTransportActions));
    }
}

在RestXxxAction中，最终会调用AbstractClient的xxx方法，并传入Action参数为XxxAction.INSTANCE，即为在ActionModule中注册的Action类

public abstract class AbstractClient implements Client {
    @Override
    public void xxx(final SuggestRequest request, final ActionListener<SuggestResponse> listener) {
        execute(XxxAction.INSTANCE, request, listener);
    }
}

在NodeClient中会根据传入的Action参数从actions获取在ActionModule中注册的TransportAction，并执行其execute()方法

public class NodeClient extends AbstractClient {
    @Override
    public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder, Client>> void execute(Action<Request, Response, RequestBuilder, Client> action, Request request, ActionListener<Response> listener) {
        headers.applyTo(request);
        TransportAction<Request, Response> transportAction = actions.get((ClientAction)action); 
        transportAction.execute(request, listener);
    }
}

在TransportAction类图个execute()方法中可以看出：
① TransportXxxAction都继承自TransportAction类
② TransportAction的execute()方法仅调用了doExecute()方法
④ TransportXxxAction重写了TransportAction的doExecute()方法
因此，在最终的对应关系为在ActionModule中注册的Action调用的为TransportXxxAction的doExecute()方法

TransportAction类图

TransportAction的execute()方法仅调用了需要子类重写的抽象方法doExecute()

public abstract class TransportAction<Request extends ActionRequest, Response extends ActionResponse> extends AbstractComponent {
    
    protected abstract void doExecute(Request request, ActionListener<Response> listener);

    public final void execute(Request request, ActionListener<Response> listener) {
        // ...
        if (filters.length == 0) {
            try {
                // TransportAction 子类都要重写这个方法
                doExecute(request, listener);
            } catch(Throwable t) {
                logger.trace("Error during transport action execution.", t);
                listener.onFailure(t);
            }
        } else {
            RequestFilterChain requestFilterChain = new RequestFilterChain<>(this, logger);
            requestFilterChain.proceed(actionName, request, listener);
        }
    }
}

3 节点间用Action匹配Handler

如果elasticsearch要请求的节点不是当前节点，则需要将请求发送到对应的节点上执行
在TransportXxxAction中会将ACTION和对应的Handler注册到对应的serverHandlers对象中，在使用transportService发送请求后，MessageChannelHandler.messageReceived()会接受到信息，然后在handleRequest()方法中获取注册的Handler，执行其messageReceived()方法或者交给线程池处理

public class TransportXxxAction {
    private final TransportService transportService;
    private final SearchService searchService;

    @Inject
    public TransportXxxAction(Settings settings, 
                        ThreadPool threadPool, 
                        TransportService transportService, 
                        ClusterService clusterService, 
                        XxxService xxxService) {
        this.transportService = transportService;
        transportService.registerHandler(XXX_ACTION_NAME, new XxxTransportHandler());
        this.xxxService= xxxService;
    }

    public void executeXxx(DiscoveryNode node, final XxxTransportRequest request, final XxxListener listener) {
        // 如果shard所在的节点id和当前节点id相同
        if (clusterService.state().nodes().localNodeId().equals(node.id())) {
            // 当前节点执行逻辑
            XxxService.execute(request);
        } else {
            // shard 所在节点不是当前节点, 发送请求执行远程节点搜索
            // node是要发送的节点
            transportService.sendRequest(node, XXX_ACTION_NAME, request, new BaseTransportResponseHandler() {
                @Override
                public QueryXxxResult newInstance() {
                    return new QueryXxxResult();
                }
                @Override
                public void handleResponse(XxxProvider response) {
                    listener.onResult(response);
                }
                @Override
                public void handleException(TransportException exp) {
                    listener.onFailure(exp);
                }
                @Override
                public String executor() {
                    return ThreadPool.Names.SAME;
                }
            });
        }
    }

    private class XxxTransportHandler extends BaseTransportRequestHandler {
        /**
         * 接收远程端口的tcp 请求, 执行Xxx 逻辑
         * @param request   TransportRequest
         * @param channel   TransportChannel
         * @throws Exception    Exception
         */
        @Override
        public void messageReceived(TransportRequest request, TransportChannel channel) throws Exception {
            // 远程节点待执行逻辑
            XxxProvider result = XxxService.execute(request);
            // 将Query结果响应发送给调用节点
            channel.sendResponse(result);
        }

        @Override
        public String executor() {
            return ThreadPool.Names.Xxx;
        }
    }
}

在transportServiceAdapter.handler()方法中，即根据注册的action获取对应的TransportHandler，如果executor是same则执行handler的messageReceived()方法，否则交给线程池执行

public class MessageChannelHandler extends SimpleChannelUpstreamHandler {
    protected String handleRequest(Channel channel, StreamInput buffer, long requestId, Version version) throws IOException {
        final String action = buffer.readString();
        transportServiceAdapter.onRequestReceived(requestId, action);
        final NettyTransportChannel transportChannel = new NettyTransportChannel(transport, transportServiceAdapter, action, channel, requestId, version, profileName);
        try {
            final TransportRequestHandler handler = transportServiceAdapter.handler(action, version);
            if (handler == null) {
                throw new ActionNotFoundTransportException(action);
            }
            final TransportRequest request = handler.newInstance();
            request.remoteAddress(new InetSocketTransportAddress((InetSocketAddress) channel.getRemoteAddress()));
            request.readFrom(buffer);
            if (ThreadPool.Names.SAME.equals(handler.executor())) {
                //noinspection unchecked
                handler.messageReceived(request, transportChannel);
            } else {
                threadPool.executor(handler.executor()).execute(new RequestHandler(handler, request, transportChannel, action));
            }
        } catch (Throwable e) {
            try {
                transportChannel.sendResponse(e);
            } catch (IOException e1) {
                logger.warn("Failed to send error message back to client for action [" + action + "]", e);
                logger.warn("Actual Exception", e1);
            }
        }
        return action;
    }
}

下面讲另外一种情况，在ActionModule中注册的Action也都会注册到handler中

public class TransportXxxAction extends HandledTransportAction {
    @Inject
    public TransportXxxAction (Settings settings, ThreadPool threadPool) {
        super(settings, XxxAction.NAME, threadPool);
}

在TransportAction类中同样使用transportService.registerHandler()方法注册handler

public abstract class HandledTransportAction<Request extends ActionRequest, Response extends ActionResponse> extends TransportAction<Request,Response>{
    protected HandledTransportAction(Settings settings, String actionName, ThreadPool threadPool, TransportService transportService, ActionFilters actionFilters){
        super(settings, actionName, threadPool, actionFilters);
        transportService.registerHandler(actionName, new TransportHandler() {
            @Override
            public Request newInstance(){
                return newRequestInstance();
            }
        });
    }
}

因此在InternalTransportClient中，主要是回调proxy.execute()

public class InternalTransportClient extends AbstractClient {
   @Inject
    public InternalTransportClient(
                                   TransportClientNodesService nodesService, 
                                   InternalTransportAdminClient,
                                   Map<String, GenericAction> actions, 
                                   Headers headers) {
        MapBuilder<Action, TransportActionNodeProxy> actionsBuilder = new MapBuilder<>();
        for (GenericAction action : actions.values()) {
            if (action instanceof Action) {
                actionsBuilder.put((Action) action, new TransportActionNodeProxy(settings, action, transportService));
            }
        }
        this.actions = actionsBuilder.immutableMap();
    }
    @Override
    public void execute(final Action action, final Request request, ActionListener listener) {
        headers.applyTo(request);
        final TransportActionNodeProxy<Request, Response> proxy = actions.get(action);
        nodesService.execute(new TransportClientNodesService.NodeListenerCallback<Response>() {
            @Override
            public void doWithNode(DiscoveryNode node, ActionListener<Response> listener) {
                proxy.execute(node, request, listener);
            }
        }, listener);
    }
}

transportService.sendRequest()中的action.name()即为Action的Name参数，已被注册到对应的handler中了

public class TransportActionNodeProxy extends AbstractComponent {
    public void execute(DiscoveryNode node, 
                            final Request request, 
                            final ActionListener<Response> listener) {
        ActionRequestValidationException validationException = request.validate();
        if (validationException != null) {
            listener.onFailure(validationException);
            return;
        }
        transportService.sendRequest(node, action.name(), 
                            request, transportOptions, 
                            new BaseTransportResponseHandler<Response>() {
            @Override
            public Response newInstance() {
                return action.newResponse();
            }

            @Override
            public String executor() {
                if (request.listenerThreaded()) {
                    return ThreadPool.Names.LISTENER;
                }
                return ThreadPool.Names.SAME;
            }

            @Override
            public void handleResponse(Response response) {
                listener.onResponse(response);
            }

            @Override
            public void handleException(TransportException exp) {
                listener.onFailure(exp);
            }
        });
    }
}