深入理解 rsync 算法：从滚动哈希到增量同步的工程实践

2026 年初，OpenBSD 团队发布了 openrsync——一个从零实现的 rsync 兼容工具，Hacker News 上引发了 250+ 热烈讨论。rsync 是每个开发者每天都在用却很少深入了解的工具：你用它部署代码、同步文件、备份数据库，但你真的理解它为什么能只传输差异部分吗？这篇文章将带你深入 rsync 的核心算法——滚动哈希（Rolling Hash）与 Delta 编码（Delta Encoding），并结合 openrsync 的设计哲学，给出可运行的代码实现和性能对比。

🔐 一、rsync 核心算法：为什么它能只传差异？

rsync 的核心思想极其优雅：在不传输完整文件的情况下，让两端达成一致。 Andrew Tridgell 在 1996 年的博士论文中提出的这个算法，至今仍是文件同步领域的基石。

1.1 经典场景：rsync 到底在解决什么问题？

假设你有一个 1GB 的日志文件，只在末尾追加了 10 行新内容。用 scp 需要传输完整 1GB，而 rsync 只传输几百字节。这个差距在大文件、弱网络环境下是决定性的。

rsync 的解决方案分为三个阶段：

1. 接收端：将已有文件切分为固定大小的块（通常 4-8KB），计算每块的弱哈希（Weak Hash）和强哈希（Strong Hash），发送给发送端
2. 发送端：用滚动哈希在新文件上滑动，逐字节匹配块哈希，找到所有匹配位置
3. 传输差异：只发送匹配不到的数据块（即"差异"部分）

📌 **记住：**rsync 的精髓在于"滚动哈希"——它能在 O(1) 时间内从上一个窗口的哈希值推算出下一个窗口的哈希值，而不是每次重新计算。这是整个算法能做到 O(n) 时间复杂度的关键。

1.2 滚动哈希（Rolling Hash）原理与实现

滚动哈希是 rsync 算法的灵魂。它的核心性质是：当窗口滑动一个字节时，可以在 O(1) 时间内更新哈希值。

rsync 使用的是 Adler-32 变体的 Rabin 指纹算法。下面是一个完整的 Python 实现：

# rsync 滚动哈希（Rolling Hash）完整实现
# 基于 Rabin 指纹算法，与 rsync 使用的 Adler-32 变体一致

class RollingHash:
    """rsync 风格的滚动哈希计算器
    
    使用两个累加器 a 和 b 来计算哈希值：
    - a = sum of all bytes in window (mod M)
    - b = sum of (window_size - i) * byte[i] (mod M)
    
    哈希值 = a + b * M
    """
    
    def __init__(self, block_size: int = 4096, modulus: int = 65521):
        """
        Args:
            block_size: 滑动窗口大小（字节），默认 4096
            modulus: 取模数，使用小于 2^16 的最大素数 65521（Adler-32 标准）
        """
        self.block_size = block_size
        self.M = modulus
        self.a = 0
        self.b = 0
        self.window = bytearray()
        self._count = 0
    
    def update(self, byte_val: int) -> int:
        """添加一个新字节到窗口，返回更新后的哈希值
        
        如果窗口未满，持续积累；窗口满后，移除最旧的字节。
        时间复杂度：O(1) —— 这就是"滚动"的含义
        """
        self.window.append(byte_val)
        self._count += 1
        
        if self._count <= self.block_size:
            # 窗口未满，持续积累
            self.a = (self.a + byte_val) % self.M
            self.b = (self.b + self.a) % self.M
        else:
            # 窗口已满，移除最旧字节，添加新字节
            old_byte = self.window.pop(0)
            self.a = (self.a - old_byte + byte_val) % self.M
            self.b = (self.b - self.block_size * old_byte + self.a) % self.M
        
        return self.a + self.b * self.M
    
    def hash(self) -> int:
        """返回当前窗口的哈希值"""
        return self.a + self.b * self.M
    
    def reset(self):
        """重置哈希计算器"""
        self.a = 0
        self.b = 0
        self.window.clear()
        self._count = 0


# === 使用示例 ===
def find_matching_blocks(data: bytes, block_hashes: dict, block_size: int = 4096):
    """在数据中查找与已知块哈希匹配的位置
    
    Args:
        data: 新文件的数据
        block_hashes: {弱哈希: (块索引, 强哈希)} 的字典
        block_size: 块大小
    
    Returns:
        匹配列表: [(offset, block_index), ...]
    """
    rh = RollingHash(block_size)
    matches = []
    
    for i, byte_val in enumerate(data):
        weak_hash = rh.update(byte_val)
        
        if weak_hash in block_hashes:
            # 弱哈希匹配，进一步验证强哈希（SHA-256）
            block_idx, strong_hash = block_hashes[weak_hash]
            start = max(0, i - block_size + 1)
            end = i + 1
            import hashlib
            actual_strong = hashlib.sha256(data[start:end]).hexdigest()
            
            if actual_strong == strong_hash:
                matches.append((start, block_idx))
    
    return matches


# 测试
if __name__ == "__main__":
    rh = RollingHash(block_size=4)
    data = b"Hello, World! This is rsync algorithm demo."
    
    for b in data:
        h = rh.update(b)
        if rh._count >= 4:
            print(f"Position {rh._count}: hash = {h}")

1.3 为什么滚动哈希是 O(n) 而不是 O(n×k)？

传统做法是每滑动一次窗口就重新计算整个窗口的哈希，时间复杂度为 O(n×k)（n 为文件大小，k 为块大小）。滚动哈希通过数学性质将其优化到 O(n)：

旧窗口：[H, e, l, l] → hash_old = f(H, e, l, l)
新窗口：[e, l, l, o] → hash_new = hash_old - f(H) + f(o)

具体到 Adler-32：

• 移除最左边字节 H：a -= H; b -= k * H
• 添加新字节 o：a += o; b += a
• 两步操作都是 O(1)，所以整体 O(n)

⚠️ **警告：**滚动哈希的碰撞率比直接计算 SHA-256 高得多。rsync 的设计是先用弱哈希快速筛选，再用强哈希（MD4/MD5/SHA-256）二次验证，两阶段策略兼顾了速度和准确性。

🚀 二、Delta 编码与增量传输实战

理解了滚动哈希后，我们来看 rsync 如何利用它实现增量传输。这个过程的核心是 Delta 编码——只传输"差异"而非"全量"。

2.1 rsync 传输协议的完整流程

# rsync 增量同步协议的简化实现
# 展示发送端和接收端的完整交互流程

import hashlib
from dataclasses import dataclass
from enum import Enum
from typing import Optional


class BlockOp(Enum):
    """数据块操作类型"""
    LITERAL = "literal"   # 原始数据，需要传输
    MATCH = "match"       # 匹配已有块，只需传输索引


@dataclass
class BlockSignature:
    """文件块的签名信息（接收端发送给发送端）"""
    index: int            # 块索引
    offset: int           # 在文件中的偏移量
    weak_hash: int        # 弱哈希（滚动哈希）
    strong_hash: str      # 强哈希（SHA-256）


@dataclass
class DeltaOp:
    """Delta 操作（发送端计算后发给接收端）"""
    op: BlockOp
    data: Optional[bytes] = None   # LITERAL 操作的原始数据
    block_index: Optional[int] = None  # MATCH 操作的块索引
    offset: int = 0


def compute_signatures(old_data: bytes, block_size: int = 4096) -> list[BlockSignature]:
    """接收端：计算已有文件的块签名
    
    将文件切分为 block_size 大小的块，为每块计算弱哈希和强哈希。
    这些签名会发送给发送端用于匹配。
    """
    signatures = []
    rh = RollingHash(block_size)
    
    for i in range(0, len(old_data), block_size):
        block = old_data[i:i + block_size]
        
        # 计算弱哈希（滚动哈希）
        for byte_val in block:
            weak_hash = rh.update(byte_val)
        
        # 计算强哈希（SHA-256，用于二次验证）
        strong_hash = hashlib.sha256(block).hexdigest()
        
        signatures.append(BlockSignature(
            index=i // block_size,
            offset=i,
            weak_hash=weak_hash,
            strong_hash=strong_hash
        ))
        rh.reset()
    
    return signatures


def compute_delta(new_data: bytes, signatures: list[BlockSignature], 
                  block_size: int = 4096) -> list[DeltaOp]:
    """发送端：计算新旧文件的 Delta 差异
    
    使用滚动哈希在新文件上滑动，匹配已知块，生成 Delta 操作序列。
    未匹配的部分作为 LITERAL 数据传输。
    """
    # 构建弱哈希索引（O(1) 查找）
    weak_hash_map: dict[int, BlockSignature] = {}
    for sig in signatures:
        weak_hash_map[sig.weak_hash] = sig
    
    delta_ops = []
    rh = RollingHash(block_size)
    literal_buffer = bytearray()
    i = 0
    
    while i < len(new_data):
        # 尝试匹配当前位置
        matched = False
        
        # 当窗口满时检查匹配
        if i >= block_size - 1:
            start = i - block_size + 1
            block = new_data[start:i + 1]
            
            # 计算弱哈希
            weak_hash = rh.hash()
            
            if weak_hash in weak_hash_map:
                sig = weak_hash_map[weak_hash]
                # 弱哈希命中，验证强哈希
                actual_strong = hashlib.sha256(block).hexdigest()
                
                if actual_strong == sig.strong_hash:
                    # 强哈希确认匹配！
                    # 先把之前累积的 LITERAL 数据输出
                    if literal_buffer:
                        delta_ops.append(DeltaOp(
                            op=BlockOp.LITERAL,
                            data=bytes(literal_buffer)
                        ))
                        literal_buffer.clear()
                    
                    delta_ops.append(DeltaOp(
                        op=BlockOp.MATCH,
                        block_index=sig.index,
                        offset=sig.offset
                    ))
                    matched = True
                    
                    # 跳过已匹配的块，重置哈希
                    i += 1
                    rh.reset()
                    # 将下一个块的前 block_size 个字节喂入哈希
                    for j in range(i, min(i + block_size, len(new_data))):
                        rh.update(new_data[j])
                    i = min(i + block_size - 1, len(new_data) - 1)
        
        if not matched:
            literal_buffer.append(new_data[i])
            if i >= block_size - 1:
                old_byte = new_data[i - block_size]
                # 滚动更新（简化处理）
            i += 1
            if i < len(new_data):
                rh.update(new_data[i - 1] if i > 0 else 0)
    
    # 输出剩余的 LITERAL 数据
    if literal_buffer:
        delta_ops.append(DeltaOp(
            op=BlockOp.LITERAL,
            data=bytes(literal_buffer)
        ))
    
    return delta_ops


def apply_delta(old_data: bytes, delta_ops: list[DeltaOp], 
                block_size: int = 4096) -> bytes:
    """接收端：根据 Delta 操作重建新文件
    
    MATCH 操作从旧文件中取对应块，LITERAL 操作直接使用原始数据。
    """
    result = bytearray()
    
    for op in delta_ops:
        if op.op == BlockOp.MATCH:
            block = old_data[op.offset:op.offset + block_size]
            result.extend(block)
        elif op.op == BlockOp.LITERAL:
            result.extend(op.data)
    
    return bytes(result)


# === 完整测试 ===
if __name__ == "__main__":
    BLOCK_SIZE = 64  # 小块用于演示
    
    # 模拟旧文件
    old_content = b"Line 1: Hello World\n" * 50
    # 模拟新文件（修改了第 3 行，追加了 2 行）
    new_content = old_content.replace(b"Line 1: Hello World\n" * 2, 
                                       b"Line 1: Modified!\n" * 2 + b"Line NEW: Added\n" * 2)
    
    print(f"旧文件大小: {len(old_content)} bytes")
    print(f"新文件大小: {len(new_content)} bytes")
    
    # 接收端计算签名
    sigs = compute_signatures(old_content, BLOCK_SIZE)
    print(f"块签名数量: {len(sigs)}")
    
    # 发送端计算 Delta
    delta = compute_delta(new_content, sigs, BLOCK_SIZE)
    
    literal_size = sum(len(op.data) for op in delta if op.op == BlockOp.LITERAL)
    match_count = sum(1 for op in delta if op.op == BlockOp.MATCH)
    
    print(f"Delta 操作数: {len(delta)}")
    print(f"  - MATCH 块: {match_count}")
    print(f"  - LITERAL 数据: {literal_size} bytes")
    print(f"传输节省: {(1 - literal_size / len(new_content)) * 100:.1f}%")
    
    # 验证重建正确性
    rebuilt = apply_delta(old_content, delta, BLOCK_SIZE)
    print(f"重建正确: {rebuilt == new_content}")

2.2 块大小选择的性能权衡

块大小是 rsync 最关键的调优参数。选择不当会严重影响性能：

💡 **提示：**rsync 的 --block-size 参数可以手动指定块大小。对于日志文件（追加写入），默认 4KB 通常最优；对于数据库文件（随机写入），增大块大小可以减少签名传输开销。

2.3 rsync 三次握手：完整的协议交互

实际的 rsync 协议比上面的简化版复杂得多，它使用三次文件列表交换来优化目录同步：

1. 第一次交换：两端交换完整的文件列表（路径、大小、修改时间）
2. 第二次交换：接收端发送需要更新的文件的块签名
3. 第三次交换：发送端发送 Delta 数据

这个设计的精妙之处在于：只有修改过的文件才需要计算块签名，避免了对整个目录树做昂贵的哈希计算。

💡 三、OpenBSD openrsync：安全第一的重新实现

2026 年 OpenBSD 团队发布的 openrsync 引发了广泛讨论。它不仅仅是一个"兼容实现"，更体现了 OpenBSD 一贯的安全哲学。

3.1 为什么需要 openrsync？

原始 rsync 存在几个长期问题：

⚠️ **警告：**rsync 历史上出现过多个严重安全漏洞（如 CVE-2022-29154 路径遍历漏洞）。如果你在生产环境使用 rsync，务必保持最新版本，并通过 SSH 传输而非 rsync daemon 模式。

3.2 openrsync 的安全设计

openrsync 利用了 OpenBSD 特有的安全机制：

// openrsync 的安全沙箱设计（伪代码示意）
// 展示 pledge/unveil 如何限制 rsync 的系统调用

#include <unistd.h>

int main(int argc, char *argv[]) {
    // 解析命令行参数...
    
    // pledge: 限制允许的系统调用类别
    // "rpath" - 读取文件系统
    // "wpath" - 写入文件系统  
    // "cpath" - 创建/删除文件
    // "stdio" - 标准输入输出
    // "dns"   - DNS 解析（用于 SSH 连接）
    if (pledge("rpath wpath cpath stdio dns", NULL) == -1) {
        err(1, "pledge");
    }
    
    // unveil: 限制文件系统可见范围
    // 只暴露源目录和目标目录，其余路径不可访问
    if (unveil(source_dir, "r") == -1) {  // 只读访问源
        err(1, "unveil source");
    }
    if (unveil(dest_dir, "rwc") == -1) {  // 读写创建访问目标
        err(1, "unveil dest");
    }
    if (unveil(NULL, NULL) == -1) {  // 锁定，不再接受新的 unveil
        err(1, "unveil lock");
    }
    
    // 即使 rsync 被攻破，攻击者也只能访问这两个目录
    // 无法读取 /etc/shadow、无法写入 /usr/bin、无法联网
    
    // ... 执行同步逻辑
}

3.3 实际使用对比

# openrsync 与原始 rsync 的使用对比
# 语法完全兼容，可以无缝替换

# 基本同步（两者完全相同）
rsync -avz /local/path/ user@remote:/remote/path/
openrsync -avz /local/path/ user@remote:/remote/path/

# 排除文件
rsync -avz --exclude='*.log' --exclude='.git/' ./ project/
openrsync -avz --exclude='*.log' --exclude='.git/' ./ project/

# 干运行（预览变更）
rsync -avzn /src/ /dst/
openrsync -avzn /src/ /dst/

# 限制带宽（KB/s）
rsync -avz --bwlimit=5000 /data/ backup:/data/
openrsync -avz --bwlimit=5000 /data/ backup:/data/

# 使用 SSH 密钥
rsync -avz -e "ssh -i ~/.ssh/deploy_key" ./ deploy@server:/app/
openrsync -avz -e "ssh -i ~/.ssh/deploy_key" ./ deploy@server:/app/

💡 **提示：**如果你是 macOS 或 Linux 用户，openrsync 目前仅在 OpenBSD 上默认可用。但 rsync 3.2.4+ 版本已经修复了大部分安全问题，保持更新即可。

3.4 开发者实战：构建自己的增量同步工具

理解了 rsync 算法后，你可以用不到 100 行代码构建一个简化版的增量同步工具：

# 简化版 rsync 增量同步工具
# 适用于需要自定义同步逻辑的场景（如部署系统、备份工具）

import hashlib
import os
from pathlib import Path
from dataclasses import dataclass
from typing import BinaryIO


@dataclass
class SyncPlan:
    """同步计划：记录需要传输的差异"""
    full_transfer: list[Path]      # 需要全量传输的新文件
    delta_transfer: list[tuple[Path, list[DeltaOp]]]  # 增量传输
    delete: list[Path]             # 需要删除的文件


def fast_file_hash(path: Path) -> str:
    """快速文件哈希：先比大小再比内容
    
    这是 rsync 的优化技巧之一：
    如果文件大小不同，直接判定为需要同步，避免昂贵的哈希计算。
    """
    stat = os.stat(path)
    # 使用文件大小 + 首尾 4KB 的哈希作为快速指纹
    with open(path, 'rb') as f:
        head = f.read(4096)
        f.seek(max(0, stat.st_size - 4096))
        tail = f.read(4096)
    
    fingerprint = f"{stat.st_size}:{hashlib.md5(head + tail).hexdigest()}"
    return fingerprint


def build_sync_plan(src_dir: Path, dst_dir: Path, 
                    block_size: int = 8192) -> SyncPlan:
    """构建同步计划
    
    比较源目录和目标目录，生成最优的同步策略：
    - 新文件 → 全量传输
    - 已修改文件 → 增量传输（只传差异）
    - 已删除文件 → 删除操作
    """
    plan = SyncPlan(full_transfer=[], delta_transfer=[], delete=[])
    
    # 收集源目录和目标目录的文件列表
    src_files = {f.relative_to(src_dir): f for f in src_dir.rglob('*') if f.is_file()}
    dst_files = {f.relative_to(dst_dir): f for f in dst_dir.rglob('*') if f.is_file()}
    
    for rel_path, src_path in src_files.items():
        dst_path = dst_dir / rel_path
        
        if rel_path not in dst_files:
            # 新文件，全量传输
            plan.full_transfer.append(src_path)
        else:
            # 文件存在，检查是否需要更新
            src_hash = fast_file_hash(src_path)
            dst_hash = fast_file_hash(dst_path)
            
            if src_hash != dst_hash:
                # 文件有变化，计算 Delta
                with open(dst_path, 'rb') as f:
                    old_data = f.read()
                with open(src_path, 'rb') as f:
                    new_data = f.read()
                
                sigs = compute_signatures(old_data, block_size)
                delta = compute_delta(new_data, sigs, block_size)
                
                # 只有当 Delta 比全量小时才用增量传输
                delta_size = sum(
                    len(op.data) for op in delta if op.op == BlockOp.LITERAL
                )
                if delta_size < len(new_data) * 0.8:
                    plan.delta_transfer.append((src_path, delta))
                else:
                    plan.full_transfer.append(src_path)
    
    # 检查需要删除的文件
    for rel_path in dst_files:
        if rel_path not in src_files:
            plan.delete.append(dst_dir / rel_path)
    
    return plan


def print_sync_report(plan: SyncPlan):
    """打印同步报告"""
    print("=" * 50)
    print("📊 同步计划报告")
    print("=" * 50)
    
    if plan.full_transfer:
        print(f"\n📤 全量传输 ({len(plan.full_transfer)} 个文件):")
        for f in plan.full_transfer:
            size = os.path.getsize(f)
            print(f"  - {f} ({size:,} bytes)")
    
    if plan.delta_transfer:
        print(f"\n🔄 增量传输 ({len(plan.delta_transfer)} 个文件):")
        for f, delta in plan.delta_transfer:
            total_size = os.path.getsize(f)
            delta_size = sum(len(op.data) for op in delta if op.op == BlockOp.LITERAL)
            savings = (1 - delta_size / total_size) * 100
            print(f"  - {f}: {total_size:,} → {delta_size:,} bytes (节省 {savings:.1f}%)")
    
    if plan.delete:
        print(f"\n🗑️  删除 ({len(plan.delete)} 个文件):")
        for f in plan.delete:
            print(f"  - {f}")
    
    total_full = sum(os.path.getsize(f) for f in plan.full_transfer)
    total_delta = sum(
        sum(len(op.data) for op in delta if op.op == BlockOp.LITERAL)
        for _, delta in plan.delta_transfer
    )
    print(f"\n📊 总传输量: {total_full + total_delta:,} bytes")
    print("=" * 50)


# 使用示例
if __name__ == "__main__":
    src = Path("./project/src")
    dst = Path("./project/dst")
    
    plan = build_sync_plan(src, dst)
    print_sync_report(plan)

这个简化版工具展示了 rsync 的核心思想：先比较再传输，能传差异就不传全量。 你可以将它集成到自己的部署脚本或备份工具中。

📊 四、rsync 替代方案对比

rsync 并不是唯一的文件同步工具。根据不同场景，你可能需要选择不同的方案：

⚡ **关键结论：**rsync 仍然是服务器间文件同步的首选。如果你需要同步到云存储（S3、OSS），用 rclone；如果你需要实时双向同步，用 Syncthing；如果你只是简单传一个文件，用 scp。

实用命令速查

# rsync 最常用的 10 个场景

# 1. 部署代码到生产服务器（排除开发文件）
rsync -avz --delete \
  --exclude='node_modules/' \
  --exclude='.git/' \
  --exclude='*.log' \
  ./ deploy@server:/app/

# 2. 增量备份数据库（只传输新增的 binlog）
rsync -avz --append-verify /var/lib/mysql/ backup:/data/mysql/

# 3. 同步大文件（限制带宽，避免影响业务）
rsync -avz --bwlimit=10000 --progress /data/dump.sql backup:/data/

# 4. 镜像整个网站（精确复制，删除多余文件）
rsync -avz --delete --exclude='cache/' /var/www/ backup:/var/www/

# 5. 通过非标准 SSH 端口传输
rsync -avz -e "ssh -p 2222" ./ user@host:/path/

# 6. 只同步修改过的文件（基于校验和而非时间戳）
rsync -avzc ./ user@host:/path/

# 7. 显示传输进度和速度
rsync -avz --progress --stats ./ user@host:/path/

# 8. 模拟运行（预览变更，不实际传输）
rsync -avzn --delete ./ user@host:/path/

# 9. 保留硬链接（节省目标空间）
rsync -avzH ./ user@host:/path/

# 10. 排除大于 100MB 的文件
rsync -avz --max-size=100m ./ user@host:/path/

✅ 总结与最佳实践

rsync 算法的优雅之处在于它用简单的数学工具（滚动哈希）解决了大规模文件同步的效率问题。理解其原理不仅能帮你更好地使用 rsync，还能启发你在其他领域（如数据库 WAL 同步、CDN 增量分发、Docker 层缓存）应用类似的增量思想。

核心要点：

• ✅ 理解滚动哈希：O(1) 窗口滑动是 rsync 效率的关键
• ✅ 合理选择块大小：小块高精度但内存大，大块低精度但省内存
• ✅ 先快后慢匹配：先用弱哈希 O(1) 筛选，再用强哈希验证
• ✅ 安全第一：始终通过 SSH 传输，保持 rsync 版本更新
• ✅ 善用排除规则：--exclude 避免传输不必要的文件

📌 **记住：**rsync 的增量思想是一通百通的。Docker 的分层构建、Git 的 packfile、数据库的 WAL 日志，本质上都是"只传差异"的变体。掌握了这个思维模式，你会在更多场景中找到优化空间。

相关工具推荐：

• 🔧 rsync 官方文档 — 完整的命令参考
• 🔧 openrsync 源码 — 学习安全的 C 代码实现
• 🔧 rclone — rsync 的云存储版本
• 🔧 lsyncd — 基于 rsync 的实时同步守护进程