flowchart TB
    subgraph "用户层"
        User[用户代码]
    end

    subgraph "REST Client 层"
        RestClient[RestClient<br/>REST 客户端]
        Builder[RestClientBuilder<br/>构建器]
        Request[Request<br/>请求对象]
        Response[Response<br/>响应对象]
    end

    subgraph "连接管理层"
        NodeSelector[NodeSelector<br/>节点选择器]
        NodeTuple[NodeTuple<br/>节点列表]
        BlackList[DeadHostState<br/>失败节点管理]
    end

    subgraph "HTTP 层"
        HttpClient[CloseableHttpAsyncClient<br/>Apache HttpClient]
        HttpRequest[HttpRequest<br/>HTTP 请求]
        HttpResponse[HttpResponse<br/>HTTP 响应]
    end

    subgraph "Elasticsearch Cluster"
        Node1[Node 1<br/>:9200]
        Node2[Node 2<br/>:9200]
        Node3[Node 3<br/>:9200]
    end

    User --> RestClient
    User --> Builder
    Builder --> RestClient

    RestClient --> Request
    RestClient --> Response
    RestClient --> NodeSelector
    RestClient --> NodeTuple
    RestClient --> BlackList

    RestClient --> HttpClient
    HttpClient --> HttpRequest
    HttpClient --> HttpResponse

    HttpClient --> Node1
    HttpClient --> Node2
    HttpClient --> Node3

flowchart TB
    Start[开始: performRequest] --> SelectNode[选择目标节点<br/>round-robin]
    SelectNode --> CheckDead[检查节点状态]
    CheckDead --> |死节点| Retry{是否可以重试?}
    CheckDead --> |活节点| BuildHTTP[构建 HTTP 请求]

    Retry --> |是| SelectNode
    Retry --> |否| Fail[抛出 IOException]

    BuildHTTP --> SendHTTP[发送 HTTP 请求]
    SendHTTP --> WaitResp[等待响应]
    WaitResp --> |成功| CheckStatus[检查 HTTP 状态码]
    WaitResp --> |失败| MarkDead[标记节点为死]

    MarkDead --> Retry

    CheckStatus --> |2xx/3xx| Success[返回 Response]
    CheckStatus --> |4xx/5xx| CheckRetry{是否可重试?}

    CheckRetry --> |是| SelectNode
    CheckRetry --> |否| ThrowException[抛出 ResponseException]

用户代码
  └─> RestClient.performRequest(Request)
       ├─> new InternalRequest(request)                    // 包装请求
       │    ├─> buildUri(pathPrefix, endpoint, params)    // 构建 URI
       │    ├─> createHttpRequest(method, uri, entity)    // 创建 HTTP 请求
       │    └─> setHeaders(httpRequest, headers)          // 设置请求头
       │
       ├─> nextNodes()                                     // 选择节点
       │    ├─> selectNodes(nodeTuple, blacklist, ...)    // 节点选择算法
       │    │    ├─> 分离存活节点和死亡节点
       │    │    ├─> NodeSelector.select(livingNodes)     // 节点过滤
       │    │    ├─> Collections.rotate(nodes, index)     // 轮询调度
       │    │    └─> return selectedNodes.iterator()
       │    │
       │    └─> new NodeTuple(iterator, authCache)
       │
       └─> performRequest(tuple, internalRequest, null)    // 递归重试
            ├─> request.createContextForNextAttempt(node)  // 创建请求上下文
            │    ├─> httpRequest.reset()                  // 重置 HTTP 请求
            │    ├─> HttpAsyncMethods.create(host, req)   // 创建请求生产者
            │    └─> new HttpClientContext()              // 创建 HTTP 上下文
            │
            ├─> client.execute(...).get()                 // 同步执行(阻塞等待)
            │    └─> Apache HttpAsyncClient
            │         ├─> 建立 TCP 连接
            │         ├─> 发送 HTTP 请求
            │         ├─> 接收 HTTP 响应
            │         └─> 返回 HttpResponse
            │
            ├─> convertResponse(request, node, response)   // 转换响应
            │    ├─> 检查状态码
            │    ├─> 处理 Gzip 解压
            │    ├─> new Response(requestLine, host, resp)
            │    │
            │    ├─> if (statusCode < 300) {              // 成功响应
            │    │    ├─> onResponse(node)                // 从黑名单移除
            │    │    └─> return response
            │    │    }
            │    │
            │    ├─> if (statusCode == 502/503/504) {     // 可重试错误
            │    │    ├─> onFailure(node)                 // 标记节点死亡
            │    │    └─> return responseException        // 返回异常,触发重试
            │    │    }
            │    │
            │    └─> else {                                // 其他错误
            │         ├─> onResponse(node)                // 保持节点存活
            │         └─> throw responseException         // 抛出异常,不重试
            │         }
            │
            ├─> if (responseException != null && tuple.hasNext()) {
            │    └─> return performRequest(tuple, request, exception)  // 递归重试
            │    }
            │
            └─> return response or throw exception

// 同步请求入口
public Response performRequest(Request request) throws IOException {
    InternalRequest internalRequest = new InternalRequest(request);
    return performRequest(nextNodes(), internalRequest, null);
}

// 递归重试的核心方法
private Response performRequest(
    final NodeTuple<Iterator<Node>> tuple,
    final InternalRequest request,
    Exception previousException
) throws IOException {
    // 1. 创建请求上下文(选择下一个节点)
    RequestContext context = request.createContextForNextAttempt(
        tuple.nodes.next(),
        tuple.authCache
    );

    HttpResponse httpResponse;
    try {
        // 2. 同步执行 HTTP 请求(内部是异步+阻塞)
        httpResponse = client.execute(
            context.requestProducer,
            context.asyncResponseConsumer,
            context.context,
            null
        ).get();
    } catch (Exception e) {
        // 3. 网络异常处理
        RequestLogger.logFailedRequest(logger, request.httpRequest, context.node, e);
        onFailure(context.node);  // 标记节点死亡

        Exception cause = extractAndWrapCause(e);
        addSuppressedException(previousException, cause);

        // 4. 如果可以重试且有下一个节点,递归调用
        if (isRetryableException(e) && tuple.nodes.hasNext()) {
            return performRequest(tuple, request, cause);
        }

        // 5. 无法重试,抛出异常
        if (cause instanceof IOException) {
            throw (IOException) cause;
        }
        if (cause instanceof RuntimeException) {
            throw (RuntimeException) cause;
        }
        throw new IllegalStateException(
            "unexpected exception type: must be either RuntimeException or IOException",
            cause
        );
    }

    // 6. 转换响应
    ResponseOrResponseException result = convertResponse(request, context.node, httpResponse);

    if (result.responseException == null) {
        // 成功,返回响应
        return result.response;
    }

    // 7. HTTP 错误响应,检查是否可以重试
    addSuppressedException(previousException, result.responseException);
    if (tuple.nodes.hasNext()) {
        // 有下一个节点,递归重试
        return performRequest(tuple, request, result.responseException);
    }

    // 所有节点都失败,抛出最终异常
    throw result.responseException;
}

// 节点选择算法
static Iterable<Node> selectNodes(
    NodeTuple<List<Node>> nodeTuple,
    Map<HttpHost, DeadHostState> blacklist,
    AtomicInteger lastNodeIndex,
    NodeSelector nodeSelector
) throws IOException {

    // 步骤1: 将节点分为存活节点和死亡节点
    List<Node> livingNodes = new ArrayList<>(
        Math.max(0, nodeTuple.nodes.size() - blacklist.size())
    );
    List<DeadNode> deadNodes = new ArrayList<>(blacklist.size());

    for (Node node : nodeTuple.nodes) {
        DeadHostState deadness = blacklist.get(node.getHost());

        // 节点不在黑名单,或者已到重试时间
        if (deadness == null || deadness.shallBeRetried()) {
            livingNodes.add(node);
        } else {
            deadNodes.add(new DeadNode(node, deadness));
        }
    }

    // 步骤2: 优先使用存活节点
    if (false == livingNodes.isEmpty()) {
        // 应用 NodeSelector 过滤节点
        List<Node> selectedLivingNodes = new ArrayList<>(livingNodes);
        nodeSelector.select(selectedLivingNodes);

        if (false == selectedLivingNodes.isEmpty()) {
            /*
             * 步骤3: 轮询调度
             * Collections.rotate() 实现负载均衡:
             * - distance = lastNodeIndex.getAndIncrement()
             * - 每次请求,distance 增加1
             * - 导致节点列表循环偏移,实现轮询
             */
            Collections.rotate(selectedLivingNodes, lastNodeIndex.getAndIncrement());
            return selectedLivingNodes;
        }
    }

    // 步骤4: 没有可用的存活节点,尝试复活死亡节点
    if (false == deadNodes.isEmpty()) {
        List<DeadNode> selectedDeadNodes = new ArrayList<>(deadNodes);

        // 应用 NodeSelector 过滤死亡节点
        nodeSelector.select(() -> new DeadNodeIteratorAdapter(selectedDeadNodes.iterator()));

        if (false == selectedDeadNodes.isEmpty()) {
            // 选择死亡时间最短的节点(最快到重试时间)
            return singletonList(Collections.min(selectedDeadNodes).node);
        }
    }

    // 步骤5: 无可用节点,抛出异常
    throw new IOException(
        "NodeSelector [" + nodeSelector + "] rejected all nodes, " +
        "living " + livingNodes + " and dead " + deadNodes
    );
}

初始节点列表: [Node0, Node1, Node2]
lastNodeIndex = 0

请求1:
  - index = 0
  - rotate(list, 0) -> [Node0, Node1, Node2]
  - 选择 Node0

请求2:
  - index = 1
  - rotate(list, 1) -> [Node2, Node0, Node1]
  - 选择 Node2

请求3:
  - index = 2
  - rotate(list, 2) -> [Node1, Node2, Node0]
  - 选择 Node1

请求4:
  - index = 3 (3 % 3 = 0)
  - rotate(list, 3) -> [Node0, Node1, Node2]
  - 选择 Node0

private ResponseOrResponseException convertResponse(
    InternalRequest request,
    Node node,
    HttpResponse httpResponse
) throws IOException {

    RequestLogger.logResponse(logger, request.httpRequest, node.getHost(), httpResponse);
    int statusCode = httpResponse.getStatusLine().getStatusCode();

    // 处理 Gzip 压缩响应
    HttpEntity entity = httpResponse.getEntity();
    if (entity != null) {
        Header header = entity.getContentEncoding();
        if (header != null && "gzip".equals(header.getValue())) {
            httpResponse.setEntity(new GzipDecompressingEntity(entity));
            httpResponse.removeHeaders(HTTP.CONTENT_ENCODING);
            httpResponse.removeHeaders(HTTP.CONTENT_LEN);
        }
    }

    Response response = new Response(
        request.httpRequest.getRequestLine(),
        node.getHost(),
        httpResponse
    );

    // 判断1: 成功响应 (2xx, 3xx)
    if (isSuccessfulResponse(statusCode) ||
        request.ignoreErrorCodes.contains(statusCode)) {

        onResponse(node);  // 从黑名单移除节点

        // 检查警告是否应该失败
        if (request.warningsHandler.warningsShouldFailRequest(response.getWarnings())) {
            throw new WarningFailureException(response);
        }

        return new ResponseOrResponseException(response);
    }

    // 判断2: 可重试的错误 (502, 503, 504)
    ResponseException responseException = new ResponseException(response);
    if (isRetryStatus(statusCode)) {
        onFailure(node);  // 标记节点死亡
        return new ResponseOrResponseException(responseException);
    }

    // 判断3: 不可重试的错误 (400, 401, 404等)
    // 这些错误是请求问题,不是节点问题,保持节点存活
    onResponse(node);
    throw responseException;
}

// 成功响应: statusCode < 300
private static boolean isSuccessfulResponse(int statusCode) {
    return statusCode < 300;
}

// 可重试状态码: 502, 503, 504
private static boolean isRetryStatus(int statusCode) {
    switch (statusCode) {
        case 502:  // Bad Gateway
        case 503:  // Service Unavailable
        case 504:  // Gateway Timeout
            return true;
    }
    return false;
}

// 请求成功,从黑名单移除节点
private void onResponse(Node node) {
    DeadHostState removedHost = this.blacklist.remove(node.getHost());
    if (logger.isDebugEnabled() && removedHost != null) {
        logger.debug("removed [" + node + "] from blacklist");
    }
}

// 请求失败,标记节点死亡
private void onFailure(Node node) {
    while (true) {
        // CAS 操作,确保线程安全
        DeadHostState previousDeadHostState = blacklist.putIfAbsent(
            node.getHost(),
            new DeadHostState(DeadHostState.DEFAULT_TIME_SUPPLIER)
        );

        if (previousDeadHostState == null) {
            // 首次失败,创建新的 DeadHostState
            if (logger.isDebugEnabled()) {
                logger.debug("added [" + node + "] to blacklist");
            }
            break;
        }

        // 已在黑名单,更新失败次数和重试时间
        if (blacklist.replace(
            node.getHost(),
            previousDeadHostState,
            new DeadHostState(previousDeadHostState)
        )) {
            if (logger.isDebugEnabled()) {
                logger.debug("updated [" + node + "] already in blacklist");
            }
            break;
        }
    }

    // 通知失败监听器
    failureListener.onFailure(node);
}

final class DeadHostState implements Comparable<DeadHostState> {

    private static final long MIN_CONNECTION_TIMEOUT_NANOS = TimeUnit.MINUTES.toNanos(1);
    static final long MAX_CONNECTION_TIMEOUT_NANOS = TimeUnit.MINUTES.toNanos(30);

    private final int failedAttempts;
    private final long deadUntilNanos;
    private final Supplier<Long> timeSupplier;

    // 构造函数1: 首次失败
    DeadHostState(Supplier<Long> timeSupplier) {
        this.failedAttempts = 1;
        this.deadUntilNanos = timeSupplier.get() + MIN_CONNECTION_TIMEOUT_NANOS;
        this.timeSupplier = timeSupplier;
    }

    // 构造函数2: 基于之前的失败状态
    DeadHostState(DeadHostState previousDeadHostState) {
        // 指数退避算法
        long timeoutNanos = (long) Math.min(
            MIN_CONNECTION_TIMEOUT_NANOS * 2 * Math.pow(2, previousDeadHostState.failedAttempts * 0.5 - 1),
            MAX_CONNECTION_TIMEOUT_NANOS
        );

        this.deadUntilNanos = previousDeadHostState.timeSupplier.get() + timeoutNanos;
        this.failedAttempts = previousDeadHostState.failedAttempts + 1;
        this.timeSupplier = previousDeadHostState.timeSupplier;
    }

    // 检查是否到重试时间
    boolean shallBeRetried() {
        return timeSupplier.get() - deadUntilNanos > 0;
    }

    long getDeadUntilNanos() {
        return deadUntilNanos;
    }

    int getFailedAttempts() {
        return failedAttempts;
    }
}

flowchart TB
    RequestFail[请求失败] --> MarkDead[标记节点为死]
    MarkDead --> RecordTime[记录失败时间]
    RecordTime --> CalcTimeout[计算重试超时<br/>min: 1分钟, max: 30分钟]

    CalcTimeout --> Wait[等待超时]
    Wait --> CheckTime{超时到期?}
    CheckTime --> |否| Wait
    CheckTime --> |是| Resurrect[复活节点]
    Resurrect --> NextRequest[下次请求尝试]

public Response performRequest(Request request) throws IOException

public class Response {
    public StatusLine getStatusLine();     // HTTP状态行
    public HttpEntity getEntity();         // 响应体
    public Header[] getHeaders();          // 响应头
    public HttpHost getHost();             // 处理请求的节点
}

Request request = new Request("GET", "/_cat/indices");
request.addParameter("format", "json");

try {
    Response response = restClient.performRequest(request);

    int status = response.getStatusLine().getStatusCode();
    String body = EntityUtils.toString(response.getEntity());

    System.out.println("Status: " + status);
    System.out.println("Response: " + body);
} catch (ResponseException e) {
    // HTTP错误响应
    Response response = e.getResponse();
    int statusCode = response.getStatusLine().getStatusCode();
} catch (IOException e) {
    // 网络错误
}

public Cancellable performRequestAsync(
    Request request,
    ResponseListener responseListener
)

public interface ResponseListener {
    void onSuccess(Response response);
    void onFailure(Exception exception);
}

Request request = new Request("GET", "/_cluster/health");

Cancellable cancellable = restClient.performRequestAsync(
    request,
    new ResponseListener() {
        @Override
        public void onSuccess(Response response) {
            try {
                String body = EntityUtils.toString(response.getEntity());
                System.out.println("Response: " + body);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onFailure(Exception exception) {
            System.err.println("Request failed: " + exception.getMessage());
        }
    }
);

// 可选: 取消请求
// cancellable.cancel();

public synchronized void setNodes(Collection<Node> nodes)

// 动态更新节点列表
List<Node> newNodes = Arrays.asList(
    new Node(new HttpHost("localhost", 9200)),
    new Node(new HttpHost("localhost", 9201)),
    new Node(new HttpHost("localhost", 9202))
);

restClient.setNodes(newNodes);

@Override
public void close() throws IOException

try (RestClient restClient = RestClient.builder(...).build()) {
    // 使用客户端
    restClient.performRequest(request);
} // 自动关闭

public Request(String method, String endpoint)

public void addParameter(String name, String value)

Request request = new Request("GET", "/_search");
request.addParameter("size", "10");
request.addParameter("from", "0");
request.addParameter("timeout", "5s");

public void setEntity(HttpEntity entity)

// 方式1: 使用NStringEntity
Request request = new Request("POST", "/my-index/_doc");
request.setEntity(new NStringEntity(
    "{\"field\":\"value\"}",
    ContentType.APPLICATION_JSON
));

// 方式2: 使用setJsonEntity (更简洁)
request.setJsonEntity("{\"field\":\"value\"}");

// 方式3: 使用ByteArrayEntity (二进制数据)
byte[] data = ...;
request.setEntity(new ByteArrayEntity(data));

public void setOptions(RequestOptions options)

// 创建通用选项
RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
builder.addHeader("Authorization", "Bearer token");
builder.setWarningsHandler(WarningsHandler.PERMISSIVE);
RequestOptions commonOptions = builder.build();

// 应用到请求
Request request = new Request("GET", "/_search");
request.setOptions(commonOptions);

// 自定义单个请求的选项
RequestOptions.Builder customBuilder = commonOptions.toBuilder();
customBuilder.addHeader("X-Custom-Header", "value");
request.setOptions(customBuilder);

public RestClientBuilder setDefaultHeaders(Header[] defaultHeaders)

Header[] headers = new Header[]{
    new BasicHeader("Authorization", "Bearer token"),
    new BasicHeader("X-Application", "my-app")
};

RestClient restClient = RestClient.builder(...)
    .setDefaultHeaders(headers)
    .build();

public RestClientBuilder setNodeSelector(NodeSelector nodeSelector)

RestClient restClient = RestClient.builder(...)
    .setNodeSelector(NodeSelector.SKIP_DEDICATED_MASTERS)
    .build();

public RestClientBuilder setFailureListener(FailureListener failureListener)

RestClient restClient = RestClient.builder(...)
    .setFailureListener(new FailureListener() {
        @Override
        public void onFailure(Node node) {
            System.err.println("Node failed: " + node.getHost());
        }
    })
    .build();

public RestClientBuilder setRequestConfigCallback(
    RequestConfigCallback requestConfigCallback
)

RestClient restClient = RestClient.builder(...)
    .setRequestConfigCallback(requestConfigBuilder -> {
        return requestConfigBuilder
            .setConnectTimeout(5000)           // 连接超时
            .setSocketTimeout(60000)           // Socket超时
            .setConnectionRequestTimeout(1000); // 从连接池获取连接超时
    })
    .build();

public RestClientBuilder setHttpClientConfigCallback(
    HttpClientConfigCallback httpClientConfigCallback
)

RestClient restClient = RestClient.builder(...)
    .setHttpClientConfigCallback(httpClientBuilder -> {
        return httpClientBuilder
            .setMaxConnTotal(100)                      // 最大连接数
            .setMaxConnPerRoute(20)                    // 每路由最大连接数
            .setKeepAliveStrategy((response, context) -> 60000)  // Keep-Alive
            .setDefaultIOReactorConfig(
                IOReactorConfig.custom()
                    .setIoThreadCount(Runtime.getRuntime().availableProcessors())
                    .build()
            );
    })
    .build();

public Node(HttpHost host)
public Node(HttpHost host, Set<HttpHost> boundHosts, String name, String version, Roles roles)

// 基本构造
Node node = new Node(new HttpHost("localhost", 9200, "http"));

// 完整构造
Node node = new Node(
    new HttpHost("localhost", 9200, "http"),
    Collections.singleton(new HttpHost("127.0.0.1", 9200, "http")),
    "node-1",
    "8.10.0",
    new Roles(true, true, true)  // master, data, ingest
);

public static class Builder {
    public Builder addHeader(String name, String value);
    public Builder setWarningsHandler(WarningsHandler warningsHandler);
    public Builder setHttpAsyncResponseConsumerFactory(
        HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory
    );
    public RequestOptions build();
}

// 创建自定义选项
RequestOptions options = RequestOptions.DEFAULT.toBuilder()
    .addHeader("X-Elastic-Product", "Elasticsearch")
    .setWarningsHandler(WarningsHandler.PERMISSIVE)
    .setHttpAsyncResponseConsumerFactory(
        new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(
            100 * 1024 * 1024  // 100 MB缓冲
        )
    )
    .build();

Request request = new Request("PUT", "/my-index");
request.setJsonEntity(
    "{"
    + "\"settings\": {"
    + "  \"number_of_shards\": 1,"
    + "  \"number_of_replicas\": 1"
    + "},"
    + "\"mappings\": {"
    + "  \"properties\": {"
    + "    \"field\": {\"type\": \"text\"}"
    + "  }"
    + "}"
    + "}"
);

Response response = restClient.performRequest(request);

Request request = new Request("POST", "/my-index/_doc");
request.addParameter("refresh", "true");
request.setJsonEntity("{\"field\":\"value\"}");

Response response = restClient.performRequest(request);

Request request = new Request("GET", "/my-index/_search");
request.setJsonEntity(
    "{"
    + "\"query\": {"
    + "  \"match\": {\"field\": \"value\"}"
    + "},"
    + "\"size\": 10"
    + "}"
);

Response response = restClient.performRequest(request);
String body = EntityUtils.toString(response.getEntity());

Request request = new Request("POST", "/_bulk");
request.addParameter("refresh", "true");

StringBuilder bulkBody = new StringBuilder();
bulkBody.append("{\"index\":{\"_index\":\"my-index\"}}\n");
bulkBody.append("{\"field\":\"value1\"}\n");
bulkBody.append("{\"index\":{\"_index\":\"my-index\"}}\n");
bulkBody.append("{\"field\":\"value2\"}\n");

request.setJsonEntity(bulkBody.toString());

Response response = restClient.performRequest(request);

classDiagram
    class RestClient {
        -CloseableHttpAsyncClient client
        -List~Header~ defaultHeaders
        -String pathPrefix
        -AtomicInteger lastNodeIndex
        -ConcurrentMap~HttpHost, DeadHostState~ blacklist
        -FailureListener failureListener
        -NodeSelector nodeSelector
        -NodeTuple nodeTuple
        +performRequest(Request) Response
        +performRequestAsync(Request, ResponseListener) Cancellable
        +setNodes(Collection~Node~)
        +close()
    }

    class Node {
        -HttpHost host
        -Set~HttpHost~ boundHosts
        -String name
        -String version
        -Roles roles
        +getHost() HttpHost
        +getName() String
    }

    class NodeTuple {
        -List~Node~ nodes
        -AtomicInteger index
        +next() Iterator~Node~
    }

    class DeadHostState {
        -long deadUntilNanos
        -int failedAttempts
        +getDeadUntilNanos() long
        +getFailedAttempts() int
    }

    RestClient --> Node
    RestClient --> NodeTuple
    RestClient --> DeadHostState

classDiagram
    class Request {
        -String method
        -String endpoint
        -Map~String, String~ parameters
        -HttpEntity entity
        -RequestOptions options
        +Request(String, String)
        +addParameter(String, String)
        +setEntity(HttpEntity)
        +setJsonEntity(String)
        +setOptions(RequestOptions)
    }

    class Response {
        -StatusLine statusLine
        -HttpEntity entity
        -HttpHost host
        -Header[] headers
        +getStatusLine() StatusLine
        +getEntity() HttpEntity
        +getHost() HttpHost
        +getHeaders() Header[]
    }

    class RequestOptions {
        -List~Header~ headers
        -WarningsHandler warningsHandler
        -HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory
        +DEFAULT RequestOptions
    }

classDiagram
    class Node {
        -HttpHost host
        -Set~HttpHost~ boundHosts
        -String name
        -String version
        -Roles roles
        -Map~String, List~String~~ attributes
        +getHost() HttpHost
        +getName() String
        +getRoles() Roles
    }

    class Roles {
        -boolean masterEligible
        -boolean data
        -boolean ingest
        +isMasterEligible() boolean
        +canContainData() boolean
        +isIngest() boolean
    }

    class NodeSelector {
        <<interface>>
        +select(Iterable~Node~)
        +ANY NodeSelector
        +SKIP_DEDICATED_MASTERS NodeSelector
    }

    Node --> Roles
    NodeSelector ..> Node

classDiagram
    class DeadHostState {
        -long deadUntilNanos
        -int failedAttempts
        -TimeSupplier timeSupplier
        +DeadHostState(TimeSupplier)
        +getDeadUntilNanos() long
        +getFailedAttempts() int
        +markDead(long)
        +shallBeRetried()
    }

    class TimeSupplier {
        <<interface>>
        +nanoTime() long
    }

    DeadHostState --> TimeSupplier

// 重试超时 = min(60秒 * 2^(失败次数-1), 30分钟)
long timeoutMinutes = Math.min(30, (long) Math.pow(2, failedAttempts - 1));
long deadUntil = currentTime + timeoutMinutes * 60 * 1_000_000_000L;

classDiagram
    class NodeTuple~T~ {
        -List~Node~ all
        -T data
        -AtomicInteger index
        +NodeTuple(List~Node~, T)
        +next() T
        +hasNext() boolean
    }

classDiagram
    class ResponseListener {
        <<interface>>
        +onSuccess(Response)
        +onFailure(Exception)
    }

    class ActionListener {
        <<interface>>
        +onResponse(Response)
        +onFailure(Exception)
    }

    ResponseListener <|.. ActionListener

ResponseListener listener = new ResponseListener() {
    @Override
    public void onSuccess(Response response) {
        // 处理响应
        try {
            String body = EntityUtils.toString(response.getEntity());
            // 业务逻辑
        } catch (IOException e) {
            // 错误处理
        }
    }

    @Override
    public void onFailure(Exception exception) {
        // 处理异常
        if (exception instanceof ResponseException) {
            Response response = ((ResponseException) exception).getResponse();
            int statusCode = response.getStatusLine().getStatusCode();
        }
    }
};

classDiagram
    class FailureListener {
        <<interface>>
        +onFailure(Node)
    }

FailureListener listener = new FailureListener() {
    @Override
    public void onFailure(Node node) {
        System.err.println("Node failed: " + node.getHost());
        // 发送告警
        // 记录日志
        // 更新监控指标
    }
};

classDiagram
    class RequestConfigCallback {
        <<interface>>
        +customizeRequestConfig(Builder) Builder
    }

    class HttpClientConfigCallback {
        <<interface>>
        +customizeHttpClient(HttpAsyncClientBuilder) HttpAsyncClientBuilder
    }

// RequestConfigCallback
RequestConfigCallback requestConfig = requestConfigBuilder -> {
    return requestConfigBuilder
        .setConnectTimeout(5000)
        .setSocketTimeout(60000);
};

// HttpClientConfigCallback
HttpClientConfigCallback httpClientConfig = httpClientBuilder -> {
    return httpClientBuilder
        .setMaxConnTotal(100)
        .setMaxConnPerRoute(20)
        .setSSLContext(sslContext)
        .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
};

graph TB
    RC[RestClient] --> Client[CloseableHttpAsyncClient<br/>Apache HttpClient]
    RC --> NT[NodeTuple<br/>节点列表]
    RC --> BL[Blacklist<br/>失败节点黑名单]

    NT --> Node[Node<br/>节点信息]
    Node --> Roles[Roles<br/>节点角色]

    BL --> DHS[DeadHostState<br/>失败状态]

    RC --> Req[Request<br/>请求对象]
    RC --> Resp[Response<br/>响应对象]

    Req --> ReqOpt[RequestOptions<br/>请求选项]

    RC --> NS[NodeSelector<br/>节点选择器]
    RC --> FL[FailureListener<br/>失败监听器]
    RC --> RL[ResponseListener<br/>响应监听器]

    style RC fill:#e1f5ff
    style Client fill:#ffe1e1
    style NT fill:#e1ffe1

graph TB
    subgraph "用户应用层"
        App[应用代码]
    end

    subgraph "Client 模块"
        RC[RestClient]
        Req[Request]
        Resp[Response]
        Builder[RestClientBuilder]
    end

    subgraph "连接管理"
        NT[NodeTuple<br/>节点列表]
        NS[NodeSelector<br/>节点选择器]
        BL[Blacklist<br/>失败节点黑名单]
        DHS[DeadHostState<br/>失败状态]
    end

    subgraph "请求处理"
        IR[InternalRequest<br/>内部请求]
        RCtx[RequestContext<br/>请求上下文]
        Conv[ResponseConverter<br/>响应转换]
    end

    subgraph "Apache HttpClient"
        HAC[HttpAsyncClient]
        HP[HttpAsyncRequestProducer]
        HC[HttpAsyncResponseConsumer]
    end

    subgraph "Elasticsearch 集群"
        N1[Node 1]
        N2[Node 2]
        N3[Node 3]
    end

    App -->|1. 创建请求| Req
    App -->|2. 发送请求| RC
    RC -->|3. 包装请求| IR
    RC -->|4. 选择节点| NT
    NT -->|5. 应用过滤| NS
    NT -->|6. 检查黑名单| BL
    BL -->|7. 获取状态| DHS
    RC -->|8. 创建上下文| RCtx
    RCtx -->|9. 创建生产者| HP
    RCtx -->|10. 创建消费者| HC
    RC -->|11. 执行请求| HAC
    HAC -->|12. 发送 HTTP| N1
    HAC -->|12. 发送 HTTP| N2
    HAC -->|12. 发送 HTTP| N3
    HAC -->|13. 返回响应| RC
    RC -->|14. 转换响应| Conv
    Conv -->|15. 更新状态| BL
    RC -->|16. 返回结果| Resp
    Resp -->|17. 返回应用| App

    style RC fill:#e1f5ff
    style BL fill:#ffe1e1
    style HAC fill:#e1ffe1

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant RC as RestClient
    participant IR as InternalRequest
    participant NT as NodeTuple
    participant NS as NodeSelector
    participant BL as Blacklist<br/>(ConcurrentMap)
    participant RCtx as RequestContext
    participant HTTP as HttpAsyncClient
    participant ES as Elasticsearch Node

    User->>RC: performRequest(request)

    Note over RC: 阶段1: 请求包装

    RC->>IR: new InternalRequest(request)
    IR->>IR: buildUri(pathPrefix, endpoint, params)
    IR->>IR: createHttpRequest(method, uri, entity)
    IR->>IR: setHeaders(defaultHeaders + requestHeaders)
    IR-->>RC: internalRequest

    Note over RC,ES: 阶段2: 节点选择

    RC->>RC: nextNodes()
    RC->>NT: 获取节点列表
    NT-->>RC: nodes = [Node1, Node2, Node3]

    RC->>BL: 检查黑名单
    loop 遍历节点
        BL->>BL: deadness = blacklist.get(node)
        alt 节点不在黑名单 or 已到重试时间
            BL-->>RC: 加入 livingNodes
        else 节点在黑名单且未到重试时间
            BL-->>RC: 加入 deadNodes
        end
    end

    RC->>NS: select(livingNodes)
    NS->>NS: 过滤节点(如跳过专用主节点)
    NS-->>RC: selectedNodes

    RC->>RC: Collections.rotate(nodes, lastNodeIndex++)
    RC-->>RC: nodeIterator

    Note over RC,ES: 阶段3: 创建请求上下文

    RC->>IR: createContextForNextAttempt(node1)
    IR->>IR: httpRequest.reset()
    IR->>RCtx: new RequestContext(node1)
    RCtx->>RCtx: 创建 HttpAsyncRequestProducer
    RCtx->>RCtx: 创建 HttpAsyncResponseConsumer
    RCtx->>RCtx: 创建 HttpClientContext
    RCtx-->>IR: context
    IR-->>RC: context

    Note over RC,ES: 阶段4: 发送 HTTP 请求

    RC->>HTTP: execute(producer, consumer, context, null).get()
    Note over HTTP: 异步执行,同步等待

    HTTP->>HTTP: 从连接池获取连接
    HTTP->>ES: TCP 连接 + HTTP 请求

    ES->>ES: 处理请求<br/>(索引/搜索/集群操作等)

    ES-->>HTTP: HTTP 响应 + 状态码
    HTTP-->>RC: HttpResponse

    Note over RC,BL: 阶段5: 响应处理

    RC->>RC: convertResponse(request, node1, httpResponse)

    alt 响应体是 Gzip 压缩
        RC->>RC: 解压缩 Gzip
        RC->>RC: 移除 Content-Encoding 头
    end

    RC->>RC: new Response(requestLine, host, httpResponse)
    RC->>RC: 检查 HTTP 状态码

    alt statusCode < 300 (成功)
        RC->>BL: onResponse(node1)
        BL->>BL: blacklist.remove(node1.host)
        Note over BL: 从黑名单移除节点

        alt 有警告且配置为失败
            RC-->>User: throw WarningFailureException
        else 正常成功
            RC-->>User: return Response
        end

    else statusCode = 502/503/504 (可重试)
        RC->>BL: onFailure(node1)

        BL->>BL: previousState = blacklist.get(node1.host)

        alt 首次失败
            BL->>BL: new DeadHostState()<br/>deadUntil = now + 1分钟
            BL->>BL: blacklist.put(node1.host, state)
        else 再次失败
            BL->>BL: new DeadHostState(previousState)<br/>failedAttempts++<br/>deadUntil = now + 指数退避时间
            BL->>BL: blacklist.replace(node1.host, newState)
        end

        Note over RC: 有其他节点,递归重试

        RC->>RC: nextNodes() → 选择 node2
        RC->>HTTP: execute(...) → node2

    else statusCode = 400/401/403/404 (请求错误)
        RC->>BL: onResponse(node1)
        BL->>BL: blacklist.remove(node1.host)
        Note over BL: 保持节点存活(错误是请求问题)

        RC-->>User: throw ResponseException(response)
    end

    Note over RC,ES: 阶段6: 异常处理

    alt 网络异常 (IOException, ConnectTimeout, SocketTimeout)
        HTTP-->>RC: Exception
        RC->>RC: RequestLogger.logFailedRequest()
        RC->>BL: onFailure(node1)
        BL->>BL: 标记节点死亡

        alt 有其他节点可重试
            RC->>RC: 递归调用 performRequest(tuple, request)
            Note over RC: 选择下一个节点重试
        else 所有节点都失败
            RC-->>User: throw IOException<br/>(包含所有失败的 suppressed异常)
        end
    end

// InternalRequest 构造函数
URI uri = buildUri(pathPrefix, request.getEndpoint(), params);
this.httpRequest = createHttpRequest(request.getMethod(), uri, request.getEntity(), compressionEnabled);
setHeaders(httpRequest, request.getOptions().getHeaders());

// 节点选择算法
List<Node> livingNodes = new ArrayList<>();
List<DeadNode> deadNodes = new ArrayList<>();

for (Node node : nodeTuple.nodes) {
    DeadHostState deadness = blacklist.get(node.getHost());
    if (deadness == null || deadness.shallBeRetried()) {
        livingNodes.add(node);
    } else {
        deadNodes.add(new DeadNode(node, deadness));
    }
}

nodeSelector.select(selectedLivingNodes);
Collections.rotate(selectedLivingNodes, lastNodeIndex.getAndIncrement());