如何用kotlinx.serialization将CBOR编码数据写入OutputStream或预分配ByteBuffer？

方案一：借助kotlinx-io实现流式编码（推荐） #

kotlinx.serialization与kotlinx-io库集成良好，可通过StreamingBinaryEncoder将CBOR编码数据直接写入OutputStream或预分配的ByteBuffer，避免一次性分配大内存块。

1. 添加kotlinx-io依赖（需与kotlinx-serialization-cbor版本兼容）：

<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-io-jvm</artifactId>
</dependency>

2. 写入OutputStream的示例：

import kotlinx.io.buffered
import kotlinx.serialization.cbor.Cbor
import kotlinx.serialization.encodeTo
import kotlinx.serialization.protobuf.StreamingBinaryEncoder
import java.io.FileOutputStream

fun <T> encodeCborToStream(data: T, serializer: KSerializer<T>, outputStream: OutputStream) {
    val cbor = Cbor.Default
    // 用buffered包装OutputStream提升写入效率
    outputStream.buffered().use { sink ->
        cbor.encodeTo(StreamingBinaryEncoder(sink), serializer, data)
    }
}

// 调用示例
val largeData = YourLargeDataClass(...)
FileOutputStream("output.cbor").use { stream ->
    encodeCborToStream(largeData, YourLargeDataClass.serializer(), stream)
}

3. 写入预分配ByteBuffer的示例：

import kotlinx.io.ByteBufferSink
import kotlinx.serialization.cbor.Cbor
import kotlinx.serialization.encodeTo
import kotlinx.serialization.protobuf.StreamingBinaryEncoder
import java.nio.ByteBuffer

fun <T> encodeCborToByteBuffer(data: T, serializer: KSerializer<T>, buffer: ByteBuffer) {
    val cbor = Cbor.Default
    val sink = ByteBufferSink(buffer)
    cbor.encodeTo(StreamingBinaryEncoder(sink), serializer, data)
    // 编码完成后翻转缓冲，方便后续读取
    buffer.flip()
}

// 调用示例
val preAllocatedBuffer = ByteBuffer.allocate(1024 * 1024) // 预分配1MB缓冲
val largeData = YourLargeDataClass(...)
encodeCborToByteBuffer(largeData, YourLargeDataClass.serializer(), preAllocatedBuffer)

方案二：自定义BinaryEncoder实现流式写入（无额外依赖） #

若不想引入kotlinx-io，可自定义BinaryEncoder，通过预分配的字节数组作为缓冲，将编码数据逐段写入OutputStream：

import kotlinx.serialization.cbor.Cbor
import kotlinx.serialization.encoding.BinaryEncoder
import kotlinx.serialization.encoding.CompositeEncoder
import kotlinx.serialization.serializer
import java.io.OutputStream

class OutputStreamEncoder(
    private val outputStream: OutputStream,
    private val buffer: ByteArray = ByteArray(8192), // 可根据需求调整缓冲大小
    private val delegate: BinaryEncoder = Cbor.Default.encoder
) : BinaryEncoder by delegate {
    private var bufferPos = 0

    override fun encodeByte(value: Byte) {
        if (bufferPos >= buffer.size) flush()
        buffer[bufferPos++] = value
    }

    override fun encodeByteArray(value: ByteArray) {
        val remaining = buffer.size - bufferPos
        if (value.size <= remaining) {
            value.copyInto(buffer, bufferPos)
            bufferPos += value.size
        } else {
            flush()
            outputStream.write(value)
        }
    }

    override fun encodeShort(value: Short) {
        if (bufferPos + 2 > buffer.size) flush()
        buffer[bufferPos++] = (value.toInt() shr 8).toByte()
        buffer[bufferPos++] = value.toByte()
    }

    override fun encodeInt(value: Int) {
        if (bufferPos + 4 > buffer.size) flush()
        buffer[bufferPos++] = (value shr 24).toByte()
        buffer[bufferPos++] = (value shr 16).toByte()
        buffer[bufferPos++] = (value shr 8).toByte()
        buffer[bufferPos++] = value.toByte()
    }

    override fun encodeLong(value: Long) {
        if (bufferPos + 8 > buffer.size) flush()
        buffer[bufferPos++] = (value shr 56).toByte()
        buffer[bufferPos++] = (value shr 48).toByte()
        buffer[bufferPos++] = (value shr 40).toByte()
        buffer[bufferPos++] = (value shr 32).toByte()
        buffer[bufferPos++] = (value shr 24).toByte()
        buffer[bufferPos++] = (value shr 16).toByte()
        buffer[bufferPos++] = (value shr 8).toByte()
        buffer[bufferPos++] = value.toByte()
    }

    override fun beginStructure(descriptor: SerialDescriptor): CompositeEncoder {
        // 复合类型开始前，若缓冲占用过半则提前刷新
        if (bufferPos > buffer.size / 2) flush()
        return delegate.beginStructure(descriptor)
    }

    fun flush() {
        if (bufferPos > 0) {
            outputStream.write(buffer, 0, bufferPos)
            bufferPos = 0
        }
    }
}

// 使用示例
fun <T> encodeCborToStream(data: T, serializer: KSerializer<T>, outputStream: OutputStream) {
    val encoder = OutputStreamEncoder(outputStream)
    try {
        Cbor.Default.encodeTo(encoder, serializer, data)
        encoder.flush()
    } finally {
        outputStream.close()
    }
}

注意事项 #

• StreamingBinaryEncoder虽为Protobuf设计，但因CBOR同样遵循BinaryEncoder接口规范，可直接复用。
• 自定义Encoder时，需覆盖所有基础类型的encode方法，避免数据写入不完整。
• 预分配ByteBuffer时，建议根据预估数据大小设置容量，减少扩容开销。

内容的提问来源于stack exchange，提问作者Андрей Щеглов