Java WebSocket.Listener 怎样控制下一条消息的接收节奏

Java HttpClient 的 WebSocket 客户端并不会因为你实现了 WebSocket.Listener 就自动按业务处理速度接收消息。真正控制“还允许调用多少次接收回调”的接口是 WebSocket.request(long n)：它增加内部接收计数器，每次调用 onText、onBinary、onPing、onPong 或 onClose 时消耗一次。

官方地址：https://docs.oracle.com/en/java/javase/25/docs/api/java.net.http/java/net/http/WebSocket.Listener.html

如果每条回调都需要较重的解析、落库或排队，优先在处理完成后调用 request(1)；如果处理很轻且可以接受一小段缓冲，再申请一个有限批次。CompletionStage 只表示回调数据何时可以被 WebSocket 回收，不能替代 request(n) 的接收许可。

下面按一个“服务端持续推送、客户端处理能力有限”的场景拆开这几个容易混淆的概念。示例只聚焦 JDK WebSocket Listener 的接收节奏，不展开分片消息拼接、重连或第三方响应式库。

先把“消息节奏”拆成三个不同问题

我第一次遇到这类问题时，最容易把三件事混成一句“控制消息速度”：允许 WebSocket 调用几次接收方法、一次回调是不是完整业务消息，以及回调参数什么时候可以安全释放。它们分别由 request(n)、last 和返回的 CompletionStage 参与表达。

Oracle 文档明确说明，接收回调之间按顺序调用，但接收方法返回的 CompletionStage 与回调次数计数器没有直接关系。也就是说，不能只返回一个“尚未完成”的 future，就把 request(n) 省掉。

request(n) 控制的是回调次数，不是完整消息数

新建的 WebSocket 接收计数器为零；在计数器为零时，WebSocket 不会调用接收方法。调用 request(n) 后，计数器增加 n，每次收到一次接收方法调用就减一。n 必须大于零，否则会抛出 IllegalArgumentException。

文本和二进制消息尤其要注意 last=false：这表示当前回调之后仍有同一条消息的内容。此时 request(1) 只申请下一次回调，不等于申请下一条完整业务消息。只有你在 last=true 的回调中看到末段，才能把已经累积的片段交给业务层。

import java.net.http.WebSocket;
import java.util.concurrent.CompletionStage;

// 用最小 Listener 展示“每次只申请一个接收回调”的节奏控制。
final class OneAtATimeListener implements WebSocket.Listener {
    @Override
    public void onOpen(WebSocket webSocket) {
        // 建立连接后先申请一次；不申请就不会进入 onText/onBinary。
        webSocket.request(1);
    }

    @Override
    public CompletionStage> onText(
            WebSocket webSocket, CharSequence data, boolean last) {
        // data 可能只是完整文本消息的一段，last 才说明是否到达末段。
        System.out.println("收到文本片段：" + data + "，last=" + last);

        // 这里可以把 data 交给自己的累积器；不要把它直接当成完整消息。
        webSocket.request(1);

        // null 表示本次 CharSequence 可以立即被实现回收。
        return null;
    }

    @Override
    public void onError(WebSocket webSocket, Throwable error) {
        // 出错后输入输出都已关闭，不要再次 request。
        System.err.println("WebSocket 接收失败：" + error.getMessage());
    }
}

这个写法适合“每个片段都要经过限速队列或成本较高的判断”的场景。若业务必须按完整消息处理，就把片段追加到缓冲区，在 last 为 true 时提交业务任务，再申请下一次回调；这样控制的是回调节奏，而不是假设服务端一次发送就一定对应一次回调。

单次申请和批量申请怎么选

选择 request(1) 还是 request(8)、request(32)，本质上是“处理延迟”和“吞吐量”之间的工程取舍。不要把较大的 n 当成性能优化的默认答案。

• 单次申请：适合解析慢、需要背压到外部队列、每个片段都要做权限或大小判断的接收器。处理完当前回调，再申请下一次，边界最清楚。
• 有限批量：适合回调处理很轻、能接受少量已进入回调但尚未完成业务处理的数据。批量值应固定且可解释，避免用极大的数制造不可控积压。
• 近似无限：只适合确实有独立缓冲和稳定消费能力的场景。它会削弱 WebSocket 层的需求边界，不能代替业务队列的容量控制。

我的经验是先从 request(1) 写出正确的生命周期，再根据可观测的排队长度、处理耗时和内存上限增加一个小批量。这样调优时改变的是一个明确的参数，不会同时改变消息拼接和异常处理逻辑。

CompletionStage 不是第二个 request

onText 和 onBinary 返回的 CompletionStage 主要用于表示 WebSocket 何时可以回收传给 Listener 的 CharSequence 或 ByteBuffer。如果异步任务还在使用这段数据，就应该先复制出业务所需内容，或返回一个在使用结束后完成的 stage；stage 完成后不要再访问原始参数。

import java.net.http.WebSocket;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

// 异步处理前复制文本，避免任务结束前继续依赖 Listener 的原始参数。
final class AsyncTextListener implements WebSocket.Listener {
    @Override
    public void onOpen(WebSocket webSocket) {
        // 先给出一个明确的接收许可，后续在回调中续租。
        webSocket.request(1);
    }

    @Override
    public CompletionStage> onText(
            WebSocket webSocket, CharSequence data, boolean last) {
        // 转成独立 String，异步任务不再持有实现管理的 CharSequence。
        String ownedText = data.toString();
        CompletableFuture handled = CompletableFuture.runAsync(() -> {
            // 这里模拟把 ownedText 写入业务队列或执行解析。
            consume(ownedText, last);
        });

        // 这个 stage 表示本次数据何时处理完，不负责申请下一次回调。
        handled.whenComplete((ignored, error) -> webSocket.request(1));
        return handled;
    }

    private static void consume(String text, boolean last) {
        // last=false 只表示片段未结束，完整消息仍需由上层累积。
        System.out.println(text.length() + " chars, last=" + last);
    }
}

这里的两个动作故意分开：handled 保护异步任务使用的数据生命周期，whenComplete 中的 request(1) 决定什么时候允许下一个接收回调。如果处理失败，应先记录错误并选择取消或关闭，而不是无条件继续申请。

关闭和异常时要停止续租

onClose 是指定 WebSocket 的最后一次 Listener 调用；onError 也表示该连接的输入和输出已经关闭。两种路径都不应继续调用 request。如果业务主动放弃接收，可以调用 cancel 吗？Java HttpClient WebSocket 使用的是 abort() 关闭输入输出，和响应式流里的 Subscription.cancel() 不是同一个 API。

@Override
public CompletionStage> onClose(
        WebSocket webSocket, int statusCode, String reason) {
    // onClose 已经是该 WebSocket 的最后一次 Listener 调用。
    System.out.println("连接关闭：" + statusCode + " / " + reason);
    return null;
}

private void stopReceiving(WebSocket webSocket) {
    // 主动放弃时关闭输入输出；不要用 request(0) 表示暂停。
    webSocket.abort();
}

需要“暂停一会儿再继续”的场景，不要用 request(0)，因为 n 必须为正数。更稳妥的方式是暂时不申请下一次回调，把待处理的数据交给有容量上限的业务队列；恢复时再按一个明确的 n 调用 request(n)。如果连接已经进入关闭或错误状态，恢复动作应被状态机拒绝。

上线前的接收节奏清单

1. 连接建立后是否明确调用了第一次 request(n)？
2. 是否把 n 误写成完整业务消息数量，而忽略 last=false？
3. 业务是否会在异步任务中继续访问原始 CharSequence 或 ByteBuffer？
4. 异常、关闭和主动放弃后，是否阻止了后续的 request？
5. 批量 n 是否有容量、延迟和处理耗时依据，而不是简单设成很大？

把这五项放进代码评审，通常比单独盯着 onText 的实现更容易发现接收节奏问题。最小可行方案仍然是：连接打开时申请一次，回调按 last 判断消息边界，处理完成后申请下一次，关闭或出错后停止申请。

相关问题

request(1) 会保证一次拿到一条完整文本消息吗？

不会。它只保证再允许一次接收方法调用；文本可能分为多次 onText，要结合 last 判断是否到达完整消息末段。

CompletionStage 没有完成时还会收到下一次回调吗？

有可能。它和接收计数器是两套语义；是否收到下一次回调取决于之前是否仍有可用的 request(n) 额度。

可以用 request(0) 暂停 WebSocket 吗？

不可以，request(n) 的 n 必须大于零。需要暂停时不要继续申请新的回调，或者在明确的关闭策略下调用 abort()。