高效地将多个zlib压缩数据流合并为单个流

Question

如果我有几个用zlib压缩过的二进制字符串，有没有办法在不解压所有数据的情况下，高效地把它们合并成一个压缩字符串呢？

现在我需要做的事情的例子：

c1 = zlib.compress("The quick brown fox jumped over the lazy dog. ")
c2 = zlib.compress("We ride at dawn! ")
c = zlib.compress(zlib.decompress(c1)+zlib.decompress(c2)) # Warning: Inefficient!

d1 = zlib.decompress(c1)
d2 = zlib.decompress(c2)
d = zlib.decompress(c)

assert d1+d2 == d # This will pass!

我想要的结果的例子：

c1 = zlib.compress("The quick brown fox jumped over the lazy dog. ")
c2 = zlib.compress("We ride at dawn! ")
c = magic_zlib_add(c1+c2) # Magical method of combining compressed streams

d1 = zlib.decompress(c1)
d2 = zlib.decompress(c2)
d = zlib.decompress(c)

assert d1+d2 == d # This should pass!

我对zlib和DEFLATE算法了解不多，所以从理论上讲，这可能完全不可能。此外，我必须使用zlib；所以我不能包裹zlib，自己想出一个可以透明处理连接流的协议。

注意：如果解决方案在Python中不是很简单，我也没关系。我愿意写一些C代码，并在Python中使用ctypes。

Answer 1

我只是把@zorlak的评论变成一个答案，并加了一些代码，以便我以后能找到它。

对于第一个数据流，调用 deflate() 并使用 Z_SYNC_FLUSH。对于后续的每个数据流，调用 deflate() 也用 Z_SYNC_FLUSH，然后去掉前两个字节，接着把它们拼接在一起，同时收集 adler32 值和未压缩的长度。最后一个数据流时，调用 deflate() 用 Z_FINISH，去掉4字节的校验和，然后用所有校验和的 adler32_combine() 来替换它。

如果你能控制数据流的初始压缩过程，你可以把未压缩数据的长度、它的 Adler-32 校验和和压缩后的数据存储在某个地方。之后你就可以随意顺序拼接这些单独的数据流。

需要注意的是，我不确定这些单独的数据流是否可以有不同的压缩级别、压缩策略或窗口大小，因为 concatenate 函数会去掉除了第一个流以外的所有 zlib 头信息……

from typing import Tuple
import zlib


def prepare(data: bytes) -> Tuple[int, bytes, int]:
    deflate = zlib.compressobj()
    result = deflate.compress(data)
    result += deflate.flush(zlib.Z_SYNC_FLUSH)
    return len(data), result, zlib.adler32(data)


def concatenate(*chunks: Tuple[int, bytes, int]) -> bytes:
    if not chunks:
        return b''
    _, result, final_checksum = chunks[0]
    for length, chunk, checksum in chunks[1:]:
        result += chunk[2:]  # strip the zlib header
        final_checksum = adler32_combine(final_checksum, checksum, length)
    result += b'\x03\x00'  # insert a final empty block
    result += final_checksum.to_bytes(4, byteorder='big')
    return result


def adler32_combine(adler1: int, adler2: int, length2: int) -> int:
    # Python implementation of adler32_combine
    # The orignal C implementation is Copyright (C) 1995-2011, 2016 Mark Adler
    # see https://github.com/madler/zlib/blob/master/adler32.c#L143
    BASE = 65521
    WORD = 0xffff
    DWORD = 0xffffffff
    if adler1 < 0 or adler1 > DWORD:
        raise ValueError('adler1 must be between 0 and 2^32')
    if adler2 < 0 or adler2 > DWORD:
        raise ValueError('adler2 must be between 0 and 2^32')
    if length2 < 0:
        raise ValueError('length2 must not be negative')

    remainder = length2 % BASE
    sum1 = adler1 & WORD
    sum2 = (remainder * sum1) % BASE
    sum1 += (adler2 & WORD) + BASE - 1
    sum2 += ((adler1 >> 16) & WORD) + ((adler2 >> 16) & WORD) + BASE - remainder
    if sum1 >= BASE:
        sum1 -= BASE
    if sum1 >= BASE:
        sum1 -= BASE
    if sum2 >= (BASE << 1):
        sum2 -= (BASE << 1)
    if sum2 >= BASE:
        sum2 -= BASE

    return (sum1 | (sum2 << 16))

这里有个简单的例子：

hello = prepare(b'Hello World! ')
test = prepare(b'This is a test. ')
fox = prepare(b'The quick brown fox jumped over the lazy dog. ')
dawn = prepare(b'We ride at dawn! ')

# these all print what you would expect
print(zlib.decompress(concatenate(hello, test, fox, dawn)))
print(zlib.decompress(concatenate(dawn, fox, test, hello)))
print(zlib.decompress(concatenate(fox, hello, dawn, test)))
print(zlib.decompress(concatenate(test, dawn, hello, fox)))

Answer 2

除了需要解压第一个压缩流的gzjoin方法外，你还可以看看gzlog.h和gzlog.c这两个文件，它们可以高效地将短字符串添加到gzip文件中，而不需要每次都解压压缩流。（这个方法也可以很容易地修改成处理zlib格式的压缩数据，而不是gzip格式的。）如果你能控制第一个压缩流的创建，就可以使用这种方法。如果你不能创建第一个压缩流，那就只能用gzjoin的方法，这个方法需要解压。

这些方法都不需要重新压缩。

Answer 3

既然你愿意尝试C语言，那你可以先看看这个叫做 gzjoin 的代码。

需要注意的是，gzjoin 这个代码在合并的时候需要先解压缩，以找出需要更改的部分，但它不需要重新压缩。这其实还好，因为解压缩通常比压缩要快。