Go 使用 NATS JetStream 发送大消息，如何压缩与兜底？

2026-11-06数据阿明

结论

Go 客户端可以在发布前压缩消息，由消费者解压，以缓解大消息超过 NATS max_payload 的问题。Go 标准库的 compress/gzip 不需要额外依赖，可用于实现端到端压缩。压缩效果取决于数据内容，不能保证每条消息都能压到限制以内。文件压缩后仍超限时，可以改用 MinIO 等对象存储传输文件，并在 JetStream 消息中发送引用信息。

为什么 Stream 压缩解决不了发布限制

题目中的 --compression s2 是服务器侧的存储压缩。消息先发布到 NATS，之后才会在存储过程中压缩，所以它不会缩小发布时传给服务器的 payload，也无法绕过 max_payload 检查。

客户端压缩发生在发布之前，服务器接收的是压缩后的字节。输入材料提到，max_payload 上限为 64 MB，并建议生产环境不超过 8 MB；这些是该问题提供的背景。压缩后的 payload 仍需符合实际部署配置的限制。

Go 示例：发布前 gzip，消费时解压

下面的函数只负责转换字节数据。将 compressGZIP 的返回值传给现有的 JetStream 发布逻辑；消费者收到消息后调用 decompressGZIP。

package message

import (
	"bytes"
	"compress/gzip"
	"io"
)

func compressGZIP(src []byte) ([]byte, error) {
	var buf bytes.Buffer
	zw := gzip.NewWriter(&buf)

	if _, err := zw.Write(src); err != nil {
		_ = zw.Close()
		return nil, err
	}
	if err := zw.Close(); err != nil {
		return nil, err
	}

	return buf.Bytes(), nil
}

func decompressGZIP(src []byte) ([]byte, error) {
	zr, err := gzip.NewReader(bytes.NewReader(src))
	if err != nil {
		return nil, err
	}
	defer zr.Close()

	return io.ReadAll(zr)
}

发布前先检查压缩结果的长度，再按项目已有方式发布：

compressed, err := compressGZIP(payload)
if err != nil {
	return err
}

if len(compressed) > maxPayloadBytes {
	// 转入对象存储或分块流程
	return errPayloadTooLarge
}

// 将 compressed 交给现有的 JetStream Publish 调用。

maxPayloadBytes 应与部署配置和消息协议一致。项目需要自行定义示例中的 errPayloadTooLarge；这里没有假定具体的 NATS Go 客户端调用签名。

为超限文件准备兜底

压缩后符合限制：直接发布压缩数据。消息中应包含明确的压缩标记或格式字段，告知消费者何时解压，避免让接收端猜测内容格式。
压缩后仍然超限：将文件存入 MinIO 等对象存储，并在 JetStream 消息中发送对象标识或可访问地址，以及业务所需的元数据。访问控制、链接有效期和文件清理策略由应用定义。
必须通过消息系统传输文件：将文件切成多个块，并附上文件标识、块序号和总块数。消费者收齐并校验后再合并。还要考虑重复投递、乱序、缺块、重试和过期清理，否则可能留下不完整文件或占用存储。

注意事项

文本、JSON 等数据通常更有机会通过压缩缩小体积。已经压缩过的图片、视频和压缩包可能几乎不变，甚至变大。对于可能很大的消息，应根据压缩后的实际字节数判断能否发布，并保留超限处理路径。消费者解压时也应设置合理的输出大小限制，避免异常数据展开成过大的内容。