Scala中如何将CompressionOutputStream转换为ByteString？

针对你的需求，这里提供两种可靠的实现方式，同时纠正你之前尝试中的问题：

方法一：传统IO方式（简单直接） #

利用ByteArrayOutputStream承接CompressionOutputStream的输出，写完数据后直接转换为ByteString，适合小数据量场景：

import java.io.ByteArrayOutputStream
import java.util.zip.GZIPOutputStream // 根据实际压缩类型替换（如DeflaterOutputStream）
import akka.util.ByteString

// 示例：待压缩的原始数据
val rawData: ByteString = ByteString("需要压缩的内容")

// 创建底层字节输出流，承接压缩后的数据
val byteOut = new ByteArrayOutputStream()
// 初始化压缩输出流（替换为你实际使用的CompressionOutputStream子类）
val compressionOut = new GZIPOutputStream(byteOut)

try {
  // 将原始数据写入压缩流
  compressionOut.write(rawData.toArray)
  // 必须调用finish()确保压缩操作完成，避免数据不完整
  compressionOut.finish()
  
  // 将字节数组转换为ByteString
  val compressedBytes: ByteString = ByteString(byteOut.toByteArray)
} finally {
  // 确保资源关闭
  compressionOut.close()
  byteOut.close()
}

方法二：Akka Stream方式（适合流式大数据场景） #

如果你需要基于Akka Stream处理流式数据，修正你之前的用法：核心是用ByteArrayOutputStream作为压缩流的底层输出，待流处理完成后提取字节转换为ByteString：

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Source, StreamConverters}
import akka.util.ByteString
import java.io.ByteArrayOutputStream
import java.util.zip.GZIPOutputStream
import scala.concurrent.ExecutionContext.Implicits.global

// 初始化Akka环境
implicit val system: ActorSystem = ActorSystem("CompressionStream")

// 示例：流式原始数据（可替换为任意Source[ByteString, _]）
val rawSource = Source.single(ByteString("流式压缩测试内容"))

val byteOut = new ByteArrayOutputStream()
val compressionOut = new GZIPOutputStream(byteOut)

// 创建Sink，将流式数据写入压缩流
val compressionSink = StreamConverters.fromOutputStream(() => compressionOut)

// 运行流并处理结果
rawSource.runWith(compressionSink).onComplete { _ =>
  // 完成压缩并提取结果
  compressionOut.finish()
  val compressedBytes: ByteString = ByteString(byteOut.toByteArray)
  
  // 关闭资源并终止Akka环境
  compressionOut.close()
  byteOut.close()
  system.terminate()
}

你之前尝试中的问题说明 #

1. Akka Stream用法误区：你获取的Sink[ByteString, Future[IOResult]]是用来接收数据写入流的终端，需要将待压缩数据通过Source发送到这个Sink，待流处理完成后，从底层的ByteArrayOutputStream提取压缩后的字节转换为ByteString。
2. BufferedOutputStream错误：BufferedOutputStream只是包装其他输出流的缓冲层，没有getBytes()方法，且初始化时必须传入底层输出流。正确的做法是用ByteArrayOutputStream作为压缩流的底层接收端，它提供toByteArray()方法获取最终字节。

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