Spring Stomp over Websocket：流式传输大文件

Question

我在网页中的 SockJs 客户端发送帧大小为 16K 的消息。消息大小限制决定了我可以传输的文件的最大大小。

以下是我在文档中找到的内容。

/**
 * Configure the maximum size for an incoming sub-protocol message.
 * For example a STOMP message may be received as multiple WebSocket messages
 * or multiple HTTP POST requests when SockJS fallback options are in use.
 *
 * <p>In theory a WebSocket message can be almost unlimited in size.
 * In practice WebSocket servers impose limits on incoming message size.
 * STOMP clients for example tend to split large messages around 16K
 * boundaries. Therefore a server must be able to buffer partial content
 * and decode when enough data is received. Use this property to configure
 * the max size of the buffer to use.
 *
 * <p>The default value is 64K (i.e. 64 * 1024).
 *
 * <p><strong>NOTE</strong> that the current version 1.2 of the STOMP spec
 * does not specifically discuss how to send STOMP messages over WebSocket.
 * Version 2 of the spec will but in the mean time existing client libraries
 * have already established a practice that servers must handle.
 */
public WebSocketTransportRegistration setMessageSizeLimit(int messageSizeLimit) {
    this.messageSizeLimit = messageSizeLimit;
    return this;
}

我的问题： 我可以设置部分消息传递，以便文件逐部分传输，而不是像现在那样作为单个消息传输吗？

更新：仍在寻找具有部分消息传递的解决方案 同时现在使用 HTTP 处理大消息（这是我的应用程序中的文件上传/下载）。

Answer 1

我可以设置部分消息传递，以便文件逐个传输，而不是像现在那样作为单个消息传输吗？

是的。这是我的 Spring boot 实验项目中的相关配置 - 基本上UploadWSHandler是注册和WebSocketTransportRegistration.setMessageSizeLimit设置的。

@Configuration
@EnableWebSocket
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer implements WebSocketConfigurer {
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new UploadWSHandler(), "/binary");
    }
    
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration.setMessageSizeLimit(50 * 1024 * 1024);
    }
}

UploadWShandler 如下所示。抱歉这里代码太多 - 关键点

• supportsPartialMessage 返回true。
• handleBinaryMessage将使用部分消息多次调用，因此我们需要组装字节。因此afterConnectionEstablished，使用 websocket URL 查询建立身份。但是您不必使用这种机制。我选择这种机制的原因是保持客户端简单，以便我webSocket.send(files[0])只调用一次，即我不会在 javascript 端切片文件 blob 对象。（旁注：我想在客户端使用普通的 websocket - 没有跺脚/袜子）
• 内部客户端分块机制提供message.isLast()最后一条消息
• 出于演示目的，我将其写入文件系统并在内存中累积这些字节，FileUploadInFlight但您不必这样做，并且可以随时在其他地方流式传输。

public class UploadWSHandler extends BinaryWebSocketHandler {

    Map<WebSocketSession, FileUploadInFlight> sessionToFileMap = new WeakHashMap<>();

    @Override
    public boolean supportsPartialMessages() {
        return true;
    }

    @Override
    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
        ByteBuffer payload = message.getPayload();
        FileUploadInFlight inflightUpload = sessionToFileMap.get(session);
        if (inflightUpload == null) {
            throw new IllegalStateException("This is not expected");
        }
        inflightUpload.append(payload);

        if (message.isLast()) {
            Path basePath = Paths.get(".", "uploads", UUID.randomUUID().toString());
            Files.createDirectories(basePath);
            FileChannel channel = new FileOutputStream(
                    Paths.get(basePath.toString() ,inflightUpload.name).toFile(), false).getChannel();
            channel.write(ByteBuffer.wrap(inflightUpload.bos.toByteArray()));
            channel.close();
            session.sendMessage(new TextMessage("UPLOAD "+inflightUpload.name));
            session.close();
            sessionToFileMap.remove(session);
        }
        String response = "Upload Chunk: size "+ payload.array().length;
        System.out.println(response);

    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        sessionToFileMap.put(session, new FileUploadInFlight(session));
    }

    static class FileUploadInFlight {
        String name;
        String uniqueUploadId;
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        /**
         * Fragile constructor - beware not prod ready
         * @param session
         */
        FileUploadInFlight(WebSocketSession session) {
            String query = session.getUri().getQuery();
            String uploadSessionIdBase64 = query.split("=")[1];
            String uploadSessionId = new String(Base64Utils.decodeUrlSafe(uploadSessionIdBase64.getBytes()));
            System.out.println(uploadSessionId);
            List<String> sessionIdentifiers = Splitter.on("\\").splitToList(uploadSessionId);
            String uniqueUploadId = session.getRemoteAddress().toString()+sessionIdentifiers.get(0);
            String fileName = sessionIdentifiers.get(1);
            this.name = fileName;
            this.uniqueUploadId = uniqueUploadId;
        }
        public void append(ByteBuffer byteBuffer) throws IOException{
            bos.write(byteBuffer.array());
        }
    }
}

顺便说一句，一个工作项目也是sprint-boot-with-websocked-chunking-assembly-and-fetch in with-websocked-chunking-assembly-and-fetchbranch