如何用RxJava从流中按间隔剥离字节并提取指定长度消息

步骤1：修正字节流剥离逻辑 #

你当前用buffer(100, 102)是固定大小截取，不符合“每读取n字节跳过2字节”的通用需求。假设有效块大小为n（比如示例中的4），应该用buffer(n, n+2)提取n个有效字节并跳过后续2个分隔字节，再将这些字节展开为连续的干净字节流：

// 示例中n=4，可根据实际需求调整这个值
int blockSize = 4;
Observable<Byte> cleanByteStream = byteObservable
    .buffer(blockSize, blockSize + 2) // 取blockSize个有效字节，步长为有效字节数+分隔字节数
    .flatMap(Observable::fromIterable) // 将每个buffer的字节展开为连续流
    .map(b -> (byte) b); // 转换为Byte类型，方便后续处理

步骤2：解析消息帧（长度+内容） #

干净字节流中前4字节是消息长度，后续对应长度的字节是消息内容。我们用scan操作符维护解析状态，逐步收集字节并提取完整消息：

先定义解析状态类：

private static class ParseState {
    enum Phase { READ_LENGTH, READ_BODY }
    Phase currentPhase;
    byte[] lengthBytes;
    int lengthOffset;
    int messageLength;
    byte[] messageBytes;
    int messageOffset;

    ParseState() {
        currentPhase = Phase.READ_LENGTH;
        lengthBytes = new byte[4];
        lengthOffset = 0;
    }
}

然后实现帧解析逻辑：

Observable<byte[]> completeMessages = cleanByteStream
    .scan(new ParseState(), (state, nextByte) -> {
        if (state.currentPhase == ParseState.Phase.READ_LENGTH) {
            // 收集4字节长度字段
            state.lengthBytes[state.lengthOffset++] = nextByte;
            if (state.lengthOffset == 4) {
                // 将4字节转换为int（默认大端字节序，小端需加.order(ByteOrder.LITTLE_ENDIAN)）
                state.messageLength = ByteBuffer.wrap(state.lengthBytes).getInt();
                // 切换到收集消息体阶段
                state.currentPhase = ParseState.Phase.READ_BODY;
                state.messageBytes = new byte[state.messageLength];
                state.messageOffset = 0;
            }
        } else {
            // 收集消息体字节
            state.messageBytes[state.messageOffset++] = nextByte;
            if (state.messageOffset == state.messageLength) {
                // 消息体收集完成，重置状态准备下一个消息
                state.currentPhase = ParseState.Phase.READ_LENGTH;
                state.lengthOffset = 0;
            }
        }
        return state;
    })
    // 过滤出完整的消息体
    .filter(state -> 
        state.currentPhase == ParseState.Phase.READ_LENGTH 
        && state.messageBytes != null 
        && state.messageOffset == state.messageLength
    )
    .map(state -> state.messageBytes);

步骤3：将消息写入文件 #

最后订阅消息流，把每个完整消息写入文件：

// 初始化文件输出流，按需调整路径和参数
try (FileOutputStream fos = new FileOutputStream("output.txt")) {
    completeMessages.subscribe(
        message -> {
            fos.write(message);
            fos.flush(); // 按需刷新缓冲区
        },
        error -> {
            System.err.println("处理出错：" + error.getMessage());
            error.printStackTrace();
        },
        () -> System.out.println("所有消息处理完成")
    );
} catch (IOException e) {
    e.printStackTrace();
}

关键注意事项 #

1. 字节序问题：代码默认用大端字节序转换长度字段，若你的数据流是小端字节序，修改长度转换逻辑：

   state.messageLength = ByteBuffer.wrap(state.lengthBytes)
       .order(ByteOrder.LITTLE_ENDIAN)
       .getInt();
   

2. 不完整帧处理：流结束时若存在未完成的帧（比如长度字段只收集了3字节），代码会自动忽略，若需要报错可在onComplete回调中检查状态并抛出异常。
3. 性能优化：如果数据流极大，建议用Flowable替代Observable以支持背压，避免内存溢出。

内容的提问来源于stack exchange，提问作者chhil