Elasticsearch源码解析：HTTP请求的通用处理流程

超、凢脫俗 2023-06-25 11:26 48阅读 0赞

Elasticsearch一般对外开放Http和Tcp Trarnsport两种服务形式，我们以传统的Http请求的通用处理过程来一窥ES内部的设计架构，希望能加深ES的理解，以便更好的使用它。

Elasticsearch使用Netty作为底层数据传输的基础架构，通过绑定`9300`端口，作为集群Node间的数据传输通道，也可通过Transport Client直接向此端口发送请求，不过一般不建议；绑定`9200`端口响应客户端的Http请求，这是ES目前推荐的Client和Server间的连接方式。

#### ES针对所有请求注册响应的Action ####

Http请求的Path和Method是唯一区分的标识，ES的Server和Client间有非常多的请求方式，其为每一个请求方式都注册了一个Action，类似于SpringMVC的Controller里的开放方法：

![ES针对所有请求种类注册的Action][ES_Action]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NTI5OTE3_size_16_color_FFFFFF_t_70]

public class RestGetAction extends BaseRestHandler { 
    
        public RestGetAction(final Settings settings, final RestController controller) { 
            super(settings);
            controller.registerHandler(GET, "/{index}/{type}/{id}", this);
            controller.registerHandler(HEAD, "/{index}/{type}/{id}", this);
        }
        ......
    }

#### RestAction内预处理请求，之后调用NodeClient ####

每个Action都指定了适配的Method和Path，以此来路由各种请求的处理。这篇博客我们以`RestGetActio`为例，跟踪ES对此请求的处理过程。

ES通过Netty注册针对Http请求的ChannelHandler，同时接受Client的Http Rest请求，根据请求的Method和Path路由到相应的Action的过程我们这里就不详细叙述了，以RestGetAction为例，最终会调用如下方法：

public abstract class BaseRestHandler extends AbstractComponent implements RestHandler { 
    	......
        @Override
        public final void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception { 
            // prepare the request for execution; has the side effect of touching the request parameters
            // 一般会返回一个lambda表达式,对RestChannelConsumer做一个函数式编程,重写其 accept(channel) 方法
            final RestChannelConsumer action = prepareRequest(request, client);
    		......
            // execute the action
            action.accept(channel);
        }
    }
    
    public class RestGetAction extends BaseRestHandler { 
    	......
        @Override
        public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException { 
            final GetRequest getRequest = new GetRequest(request.param("index"), request.param("type"), request.param("id"));
            ......
    		// NodeClient 处理预处理后生成的GetRequest
            return channel -> client.get(getRequest, new RestToXContentListener<GetResponse>(channel) { 
                @Override
                protected RestStatus getStatus(final GetResponse response) { 
                    return response.isExists() ? OK : NOT_FOUND;
                }
            });
        }
    }

`BaseRestHandler` 是所有RestAction的父类，提供统一的 `handleRequest` 方法处理Http请求，但使用模板方法模式留下`prepareRequest` 方法给实际的Action实现各自的处理逻辑，返回的是一个函数式编程的匿名类对象，附加请求完成时的回调函数。

每个Action都会对Request内的数据做相应的预处理，然后生成对应的Request，比如Get请求就是GetRequest，将预处理的数据放入GetRequest中，最后提交给NodeClient处理。

#### NodeClient处理请求，转发给相应的TransportAction ####

public class NodeClient extends AbstractClient { 
    
    	static Map<String, ActionHandler<?, ?>> setupActions(List<ActionPlugin> actionPlugins) { 
            ......
            // 注册TransportAction, 针对性处理相应Action的请求
            actions.register(GetAction.INSTANCE, TransportGetAction.class);
            actions.register(SearchAction.INSTANCE, TransportSearchAction.class);
            actions.register(BulkAction.INSTANCE, TransportBulkAction.class, TransportShardBulkAction.class);
    	}
    
    	public void get(final GetRequest request, final ActionListener<GetResponse> listener) { 
    		// 不同的Http请求转发到execute时, Action实例不同, 通过Action找到对应的TransportAction
            execute(GetAction.INSTANCE, request, listener);
        }
    
        public <Request extends ActionRequest, Response extends ActionResponse
            > Task executeLocally(GenericAction<Request, Response> action, Request request, ActionListener<Response> listener) { 
            // 比如 TransportGetAction.execute(request, listener)
            return transportAction(action).execute(request, listener);
        }
    
    	/** * Get the {@link TransportAction} for an {@link Action}, throwing exceptions if the action isn't available. */
        @SuppressWarnings("unchecked")
        private <Request extends ActionRequest, Response extends ActionResponse> TransportAction<Request, Response> transportAction(
            GenericAction<Request, Response> action) { 
    
            if (actions == null) { 
                throw new IllegalStateException("NodeClient has not been initialized");
            }
            // 从所有注册的action中获取指定的TransportAction实现类, 比如 TransportBulkAction, TransportGetAction
            TransportAction<Request, Response> transportAction = actions.get(action);
            if (transportAction == null) { 
                throw new IllegalStateException("failed to find action [" + action + "] to execute");
            }
            return transportAction;
        }
    
    }

`NodeClient` 内部经过一些方法调用会将`get`请求转发到`executeLocally`，所有的Rest请求都会转发到此方法，但参数上的action不同。NodeClient会先通过action找到对应的`TransportAction`，然后使用找到的TransportAction执行请求。

TransportAction，这个是针对Tcp Transport 的请求处理Action，其作用和RestAction类似，每一个RestAction都有其对应的TransportAction，比如 `RestGetAction` >> `TransportGetAction`。ES通过RestAction接受Http请求，做相应预处理，生成对应的Request，比如GetRequest，然后提交给NodeClient，NodeClient通过RestAction找到对应的TransportAction，使用其做实际处理请求。

#### TransportAction对请求的处理 ####

TransportAction是ES提供的服务于9300端口的一些列的请求处理器，不仅用作TransportClient和Server间的请求，同时也适用于Server于Server间的数据传输。当一个Shard需要另一个Shard上的数据时，其会将自己做为一个TransportClient，发送请求至指定Shard的9300端口。

也就是说`Http请求处理的底层使用的是9300端口的Tcp请求的处理器`，统一了Http和Tcp请求的处理逻辑。

TransportAction经过一些步骤后，会调用至`TransportService#sendRequest` 方法：

public <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,final TransportRequest request,
                                                              final TransportResponseHandler<T> handler) { 
        try { 
        	// 通过Node的信息获取到Connection，集群Node间的连接在节点初始化时就会建立
            Transport.Connection connection = getConnection(node);
            sendRequest(connection, action, request, TransportRequestOptions.EMPTY, handler);
        } catch (NodeNotConnectedException ex) { 
            // the caller might not handle this so we invoke the handler
            handler.handleException(ex);
        }
    }
    
    public Transport.Connection getConnection(DiscoveryNode node) { 
        if (isLocalNode(node)) { 
        	// 请求由当前Node处理，获取当前9300的Connection
            return localNodeConnection;
        } else { 
            return transport.getConnection(node);
        }
    }

##### TransportService对请求的处理 #####

`TransportService` 先根据路由的Node，找到当前Node和目标Node的Connection，这个是Node在初始化时就创建好的长连接，当然如果目标Node就是自身的话，也会获取自身的Connection。

获取到Connection之后，发送请求，间接调用至 `TransportService#sendRequestInternal` ：

/** * RequestId生成器,每次请求自增1 */
        private final AtomicLong requestIdGenerator = new AtomicLong();
    	
    	/** * 存储 requestId >> 接收到response的handler, 或者timeoutHandler * @see {@link #onResponseReceived(long)} */
        final ConcurrentMapLong<RequestHolder> clientHandlers = ConcurrentCollections.newConcurrentMapLongWithAggressiveConcurrency();
    	
    	/** * 当前节点向其他节点发送请求 * @param <T> */
        private <T extends TransportResponse> void sendRequestInternal(final Transport.Connection connection, final String action,
                                                                       final TransportRequest request,
                                                                       final TransportRequestOptions options,
                                                                       TransportResponseHandler<T> handler) { 
            DiscoveryNode node = connection.getNode();
            // 生成当前请求ID, 默认每次递增1
            final long requestId = transport.newRequestId();
            final TimeoutHandler timeoutHandler;
            try { 
            	
                if (options.timeout() == null) { 
                    timeoutHandler = null;
                } else { 
                    timeoutHandler = new TimeoutHandler(requestId);  // 如果Reequest里指定了超时时间
                }
                Supplier<ThreadContext.StoredContext> storedContextSupplier = threadPool.getThreadContext().newRestorableContext(true);
                TransportResponseHandler<T> responseHandler = new ContextRestoreResponseHandler<>(storedContextSupplier, handler);
                // 存储requestId和ResponseHandler, 这样当收到指定RequestId的Response时, 执行相应的回调
                clientHandlers.put(requestId, new RequestHolder<>(responseHandler, connection, action, timeoutHandler));
                if (timeoutHandler != null) { 
                    assert options.timeout() != null;
                    // 定时任务timeout, 当到时间时触发timeout 操作
                    timeoutHandler.future = threadPool.schedule(options.timeout(), ThreadPool.Names.GENERIC, timeoutHandler);
                }
                // TcpTransport#sendRequest
                connection.sendRequest(requestId, action, request, options); // local node optimization happens upstream
            } catch (final Exception e) { 
                ......
            }
        }
    
    	
    // 生存当前请求的RequestId, 通过AtomicLong的递增实现
    public long newRequestId() { 
        return requestIdGenerator.incrementAndGet();
    }

在发送请求前，需要先生成一个RequestId，唯一表示此次请求，这样当Channel里收到一个requestId一致的Response时，通过`clientHandlers.remove(requestId)`，就找到了之前的回调函数，不同的Action有各自的回调实现，比如GetAction，将GetResult返回给Client；SearchAction，如果有Top设置，对比各Node结果的分值，取Top X，然后通过docID到相应的Node Get数据。

如果Request里设置了超时时间，那么通过 `threadPool.schedule(...)` 设置一个定时任务，当时间到了能触发时，表示当前请求超时，执行超时逻辑。

后面的逻辑基本是数据报文组装，Netty的Channel发送请求，这里就不详细解析了。

**以上就是Elasticsearch在服务端对Http请求处理的大致的通用流程，使用RestAction体系绑定9200端口接收Http请求，转发至针对9300端口的TransportAction，统一Rest请求和Transport请求的处理逻辑，使用Netty作为底层数据传输架构，使用递增的RequestId，保存所有的RequestId和对应的ResponseHandler，当接受到对应的RequestId的Response时回调ResponseHandler， 如果响应超时，自动触发TimeOutHandler。**

[ES_Action]: https://img-blog.csdnimg.cn/20191228212417378.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NTI5OTE3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NTI5OTE3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191228220546952.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NTI5OTE3,size_16,color_FFFFFF,t_70