Java HttpClient WebSocket 连接如何处理异步回调

用 Java HttpClient 建立 WebSocket 后“连接成功但没有回调”，通常不是异步线程失效，而是接收配额还没有增加。WebSocket 初始的接收计数为 0，必须在 onOpen 中调用 request(1)。随后，request(n) 控制的是最多触发多少次接收回调，不是完整消息数；文本消息还可能被拆成多个 onText 片段。

为什么连接建立了却收不到第一条消息

这个 API 把发送和接收分成两条独立的边。buildAsync 返回成功只说明握手完成，并不代表监听器已经获得了接收名额。新建 WebSocket 的计数是 0，onOpen 是第一个回调入口，最小写法如下：

@Override
public void onOpen(WebSocket webSocket) {
    // 连接成功后申请一次接收回调，否则计数保持为 0
    webSocket.request(1);
}

这里的“1”对应一次 onText、onBinary、onPing、onPong 或 onClose 调用。它不保证拿到一条完整文本；onText 的 last 才是当前消息是否结束的标记。

用 request(1) 把异步回调变成可控背压

如果回调里要写入队列、落盘或调用业务线程池，不要收到片段后立即无限申请下一次。把片段复制成字符串，在异步工作完成后再申请下一次，既避免继续持有 API 提供的字符序列，也让业务处理速度成为接收速度的上限。

final class TextListener implements WebSocket.Listener {
    private final Executor executor;
    private final StringBuilder message = new StringBuilder();

    TextListener(Executor executor) {
        this.executor = executor;
    }

    @Override
    public void onOpen(WebSocket webSocket) {
        // 只打开一个接收名额，后续由处理完成回调继续申请
        webSocket.request(1);
    }

    @Override
    public CompletionStage> onText(WebSocket webSocket,
                                       CharSequence data,
                                       boolean last) {
        String part = data.toString(); // 先复制，异步阶段不再依赖原始 CharSequence
        return CompletableFuture.runAsync(() -> {
            message.append(part);
            if (last) {
                // last=true 才代表一个完整文本消息已经拼完
                System.out.println("完整消息: " + message);
                message.setLength(0);
            }
        }, executor).whenComplete((ignored, error) -> {
            if (error != null) {
                // 业务处理失败时停止接收并结束连接
                webSocket.abort();
                return;
            }
            if (!webSocket.isInputClosed()) {
                // 当前片段处理完成后，才发放下一次回调配额
                webSocket.request(1);
            }
        });
    }

    @Override
    public CompletionStage> onClose(WebSocket webSocket, int code, String reason) {
        // onClose 是该连接的最后一个 Listener 接收回调
        System.out.println("关闭: " + code + " / " + reason);
        return CompletableFuture.completedFuture(null);
    }

    @Override
    public void onError(WebSocket webSocket, Throwable error) {
        // onError 到达时输入和输出都已关闭，只记录并通知上层
        error.printStackTrace();
    }
}

关键区别是：返回的 CompletionStage 表示本次数据何时可以被回收，而 request(1) 才决定还能触发几次接收回调。两者都保留，才能把“异步处理”和“接收节流”对齐。若业务允许并行消费，可以一次申请更大的配额，但必须自己限制队列长度。

关闭时怎样安排 sendClose、onClose 和 abort

sendClose(WebSocket.NORMAL_CLOSURE, "") 是有序关闭输出的一次发送操作，返回的 Future 完成表示 Close 消息已经发出；它不会立刻关闭输入，输入仍可能等对端的 Close。对端的 Close 到达后才进入 onClose，此时该连接不再继续调用其他 Listener 接收方法。

因此，正常完成可以沿着“最后一条业务消息发送完成 → sendClose → 等待 onClose”推进。遇到业务处理异常、空闲超时或对端迟迟不回应时，abort() 才是立即关闭输入和输出的兜底。不要在每个 onText 里重复调用 sendClose，也不要把 onClose 当成可以继续接收数据的通知。

相关问题

request(1) 会不会只收到一条完整消息？

不会。它只申请一次接收方法调用；文本或二进制消息可能跨多个片段，必须根据 last 累计。

onText 返回 null 可以吗？

可以，等价于立即完成的阶段，适合已经同步复制并处理完数据的场景。异步处理则应返回实际的 CompletionStage。

sendClose 后还需要 abort 吗？

正常对端会回复 Close 时不需要。只有超时、协议异常或业务不再等待输入时，才用 abort 强制结束。