java - 关于压缩库在不考虑 cpu 开销的情况下使 byte[] 尽可能小的建议？

Question

如果我接近这个错误，请纠正我，但我有一个队列服务器和一堆我在集群中运行的 java worker。我的队列中的工作单元非常小，但数量很多。到目前为止，我的基准测试和对工作人员的审查表明我获得了大约 200mb/秒。

所以我想弄清楚如何通过我的带宽获得更多工作单元。目前我的 CPU 使用率不是很高(40-50%)，因为它处理数据的速度比网络发送数据的速度快。我想通过队列获得更多工作，并且愿意通过昂贵的压缩/解压缩为此付出代价(因为现在每个内核有一半处于空闲状态)。

我已经尝试过 java LZO 和 gzip，但想知道是否有更好的东西(即使它的 cpu 更昂贵)？

更新:数据是一个字节[]。基本上队列只采用那种格式，所以我使用 ByteArrayOutputStream 将两个 int 和一个 int[] 写入 byte[] 格式。 int[] 中的值都是 0 到 100 之间的整数(或 1000，但绝大多数数字是零)。这些列表非常大，从 1000 到 10,000 个项目(同样，大多数零..int [] 中的非零数字永远不会超过 100 个)

Answer 1

听起来使用利用数据结构的自定义压缩机制可能非常有效。

首先，使用 short[](16 位数据类型)代替 int[] 将减半(!)发送的数据量，您可以这样做这是因为数字很容易在 -2^15 (-32768) 和 2^15-1 (32767) 之间。这非常容易实现。

其次，您可以使用类似于游程长度编码的方案:正数表示字面上的数字，而负数表示那么多的零(在取绝对值之后)。例如

[10, 40, 0, 0, 0, 30, 0, 100, 0, 0, 0, 0] <=> [10, 40, -3, 30, -1, 100, -4]

仅用 short 替换 int 更难实现，但在最坏的情况下(1000 个数字，100 个非零，没有一个是连续的)。

我刚刚做了一些模拟来计算压缩率。我测试了上面描述的方法，以及 Louis Wasserman 和 sbridges 建议的方法。两者表现都非常好。

假设数组的长度和非零数的个数都在它们的边界之间，这两种方法都节省了大约 5400 个int(或short)平均压缩大小约为原始大小的 2.5%! run-length encoding 方法似乎节省了大约 1 个额外的 int(或平均压缩大小小 0.03%)，即基本上没有区别，所以你应该使用最容易实现的那个。以下是 50000 个随机样本的压缩率直方图(它们非常相似!)。

总结:使用short而不是int和其中一种压缩方法，您将能够将数据压缩到大约原始大小的 1%!

对于模拟，我使用了以下 R 脚本:

SIZE <- 50000

lengths <- sample(1000:10000, SIZE, replace=T)
nonzeros <- sample(1:100, SIZE, replace=T)

f.rle <- function(len, nonzero) {
  indexes <- sort(c(0,sample(1:len, nonzero, F)))
  steps <- diff(indexes)
  sum(steps > 1) + nonzero # one short per run of zeros, and one per zero
}

f.index <- function(len, nonzero) {
  nonzero * 2
}

# using the [value, -1 * number of zeros,...] method
rle.comprs <- mapply(f.rle, lengths, nonzeros)
print(mean(lengths - rle.comprs)) # average number of shorts saved

rle.ratios <- rle.comprs / lengths * 100
print(mean(rle.ratios)) # average compression ratio

# using the [(index, value),...] method
index.comprs <- mapply(f.index, lengths, nonzeros)
print(mean(lengths - index.comprs)) # average number of shorts saved

index.ratios <- index.comprs / lengths * 100
print(mean(index.ratios)) # average compression ratio


par(mfrow=c(2,1))
hist(rle.ratios, breaks=100, freq=F, xlab="Compression ratio (%)", main="Run length encoding")
hist(index.ratios, breaks=100, freq=F, xlab="Compression ratio (%)", main="Store indices")