Spring通过Websocket传输大文件时的Stomp问题

Question

Spring通过Websocket传输大文件时的Stomp问题

3

我在网页中使用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进行大型消息（即文件上传/下载）。

- kukkuz

我来到这里是因为我正在尝试自己解决这个问题。我有你要找的解决方案，但在服务器端，我不想逐部分接收，而是希望使用流（仍然是逐部分的，但使用内部缓冲区）。想知道你是否找到了更好的解决方案或者至少需要这个解决方案。 - bhantol

我现在只使用HTTP进行文件传输，如果您有通过Websockets的stream解决方案，那就太好了。 - kukkuz

我已经发布了答案，虽然我的直接动机是减少上传大文件时内存峰值的单页应用程序的测量要求，但我链接的实验项目充其量只是一个 PoC，非常原始。 - bhantol

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- bhantol · Accepted Answer

我可以设置部分消息传输吗？目前文件是作为单个消息传输的。

可以。以下是我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);
    }
}

上传WShandler如下所示。抱歉，这里有太多的代码-重点如下：

- supportsPartialMessage返回true。 - handleBinaryMessage将被多次调用以获取部分消息，因此我们需要组装字节。因此，在建立连接后，afterConnectionEstablished使用websocket URL查询建立身份识别。但您不必使用此机制。我选择此机制的原因是为了保持客户端简单，这样我只需要调用webSocket.send(files[0])一次，即不在javascript端对文件blob对象进行分片。（旁边的一点：我想在客户端上使用普通websocket-没有stomp / socks） - 内部客户端分块机制提供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());
        }
    }
}

顺便提一下，一个正在运行的项目也是在with-websocked-chunking-assembly-and-fetch分支中的spring-boot-with-websocked-chunking-assembly-and-fetch