Support alternative compressors for enhanced throughput / compression ratios
Feature Request: Plug-and-Play Job Payload Compression (Support for zstd / Custom Compressors)
Hi RQ maintainers, thanks for building and maintaining such a solid library!
Currently, RQ hardcodes zlib for payload compression and decompression in rq/job.py. For applications handling high job volumes or large payload sizes, alternative compression algorithms—such as zstandard (zstd)—offer significantly higher throughput with lower CPU overhead and better compression ratios.
Proposed Solution
Decouple zlib by introducing a Compressor protocol or pluggable interface on the Job class (or via job_class overrides). This allows users to supply custom compression strategies (e.g., zstd, lz4, or even a trained dictionary compressor) without altering core RQ internals.
An example abstract interface and implementation:
from typing import Any, ClassVar, Protocol
import zlib
import zstandard as zstd
class Compressor(Protocol):
error: ClassVar[tuple[type[Exception], ...]]
@staticmethod
def compress(data: bytes) -> bytes: ...
@staticmethod
def decompress(data: bytes) -> bytes: ...
class ZlibCompressor:
error = (zlib.error,)
compress = zlib.compress
decompress = zlib.decompress
class ZstdCompressor:
_cctx = zstd.ZstdCompressor()
_dctx = zstd.ZstdDecompressor()
error = (zstd.ZstdError,)
compress = _cctx.compress
decompress = _dctx.decompress
class Job:
compressor: type[Compressor] = ZlibCompressor
...
def to_dict(self, include_meta: bool = True, include_result: bool = True) -> dict:
...
obj: dict[str, Any] = {
...,
'data': self.compressor.compress(self.data),
...
}
def restore(self, raw_data) -> Any:
...
try:
self.data = self.compressor.decompress(raw_data)
except self.compressor.error:
self.data = raw_data
...
Thanks a lot for consideration, let me know if it's possible I can open a PR
Source: rq/rq