当前位置：首页 > article >正文

Spring WebFlux 中 WebSocket 使用 DataBuffer 的注意事项

article 2025/3/9 9:50:06

以下是修改后的完整文档，包含在多个多线程环境中使用 retain() 和 release() 方法的示例，且确保在 finally 块中调用 release()：

在 Spring WebFlux 中，WebSocketMessage 主要用于表示 WebSocket 的消息载体，其中 getPayload() 方法返回 DataBuffer，用于处理二进制数据流。在使用 DataBuffer 时，需要注意其一次性读取特性，以及潜在的内存管理问题。本文将介绍如何正确使用 DataBuffer，避免重复读取和内存泄漏。

1. 避免重复读取 DataBuffer

DataBuffer 设计为一次性读取流数据，因此，一旦被消费，后续读取将无法获取数据。例如：

String firstRead = webSocketMessage.getPayload().toString(StandardCharsets.UTF_8);
String secondRead = webSocketMessage.getPayload().toString(StandardCharsets.UTF_8); // 此处读取会失败

解决方案

如果需要多次使用 DataBuffer 的数据，可以在第一次读取时缓存：

DataBuffer dataBuffer = webSocketMessage.getPayload();
byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);
String payload = new String(bytes, StandardCharsets.UTF_8);

这样，后续可以安全地使用 payload 变量，而不会影响 DataBuffer。

2. 避免阻塞操作

Spring WebFlux 是基于响应式编程的，WebSocket 处理也应保持非阻塞。如果在 DataBuffer 处理中引入了阻塞操作（如同步 I/O 或 Thread.sleep()），可能会导致 Reactor 线程阻塞，影响整体吞吐量。

解决方案

使用 Flux/Mono 进行异步处理，例如：

session.receive()
    .map(WebSocketMessage::getPayloadAsText)  // 避免直接操作 DataBuffer
    .flatMap(payload -> processMessage(payload))
    .subscribe();

3. 处理 DataBuffer 可能带来的内存泄漏

Spring WebFlux 采用 Netty 作为默认底层引擎，而 Netty 的 ByteBuf 需要手动释放，否则可能导致内存泄漏。Spring 提供了 DataBufferUtils.release() 方法来避免 DataBuffer 占用资源不被回收。

正确的释放方式

session.receive()
    .doOnNext(message -> {
        try {
            String data = message.getPayloadAsText();
            System.out.println("Received: " + data);
        } finally {
            DataBufferUtils.release(message.getPayload());
        }
    })
    .subscribe();

DataBufferUtils.release() 仅在手动管理 DataBuffer 生命周期时才需要，如果直接通过 WebSocketMessage.getPayloadAsText() 处理字符串，不必显式释放。

4. 在 Flux/Mono 组合操作时避免数据丢失

如果 DataBuffer 被 map() 操作多次消费，可能导致数据丢失或 DataBuffer 为空。例如：

session.receive()
    .map(message -> {
        DataBuffer payload = message.getPayload();
        DataBufferUtils.release(payload); // 这里释放后，后续的 map() 操作会读取不到数据
        return payload;
    })
    .map(buffer -> buffer.toString(StandardCharsets.UTF_8)) // 这里可能会失败
    .subscribe();

正确的方式

确保 DataBuffer 只在最终消费时释放。
处理 DataBuffer 时，转换为 byte[] 以避免流式数据的重复读取。

session.receive()
    .map(WebSocketMessage::getPayload)
    .map(dataBuffer -> {
        byte[] bytes = new byte[dataBuffer.readableByteCount()];
        dataBuffer.read(bytes);
        DataBufferUtils.release(dataBuffer);  // 读取完毕后释放
        return new String(bytes, StandardCharsets.UTF_8);
    })
    .subscribe(System.out::println);

5. `retain()` 和 `release()` 方法的补充

Spring WebFlux 中，WebSocketMessage 还提供了 retain() 和 release() 方法，用于管理 DataBuffer 的引用计数和释放资源。下面介绍如何在多线程环境中正确使用这些方法。

retain() 方法

retain() 方法确保 DataBuffer 的引用计数增加，以便在需要时能够安全使用：

public WebSocketMessage retain() {
    if (reactorNetty2Present) {
        return ReactorNetty2Helper.retain(this);
    }
    DataBufferUtils.retain(this.payload);
    return this;
}

retain() 方法会增加 DataBuffer 的引用计数，防止在处理过程中被提前释放。这对于需要多个组件共享同一 DataBuffer 实例的情况非常重要。

release() 方法

release() 方法用于释放 DataBuffer，减少引用计数，释放底层资源，防止内存泄漏：

public void release() {
    DataBufferUtils.release(this.payload);
}

release() 方法通常在处理完成后调用，确保底层的 DataBuffer 被正确释放。

使用示例：在多线程环境中使用 retain() 和 release()

在 WebSocket 消息处理时，确保在多线程环境中正确管理 DataBuffer 的生命周期。示例如下，使用 retain() 保证资源被正确引用，并在 finally 块中调用 release() 确保即使出现异常时也会释放资源：

session.receive()
    .doOnNext(message -> {
        // 在多线程环境中保留引用
        message.retain();
        try {
            String data = message.getPayloadAsText();
            System.out.println("Received: " + data);
            
            // 模拟处理过程，可能会涉及多线程操作
            // 例如：通过某个线程池处理消息
            processMessageAsync(data);

        } finally {
            // 确保释放资源
            message.release();  // 释放资源
        }
    })
    .subscribe();

在上面的示例中，retain() 确保了 DataBuffer 在多个线程中可以安全访问，直到最终的 release() 被调用来释放资源。无论操作成功与否，finally 块中的 release() 都会被执行，确保不会发生内存泄漏。

6. 总结

在 Spring WebFlux 中使用 WebSocketMessage 和 DataBuffer 需要注意以下几点：

避免重复读取 DataBuffer，建议在读取后缓存数据。
避免阻塞操作，尽量使用 Flux/Mono 进行异步处理。
防止内存泄漏，在手动管理 DataBuffer 生命周期时使用 DataBufferUtils.release() 释放资源。
确保 DataBuffer 只在最终消费时释放，避免 Flux 流程中数据丢失。
使用 retain() 和 release() 方法 来管理 DataBuffer 的引用计数，确保资源的正确释放，特别是在多线程环境中，确保在 finally 中释放资源。