分布式ID生成器设计全攻略：从雪花算法到号段模式（PHP实战篇）

0 - 0000000000 0000000000 0000000000 0000000000 0 - 00000 - 00000 - 000000000000
|   <----------- 41位毫秒时间戳 -----------> |  <10位机器> <12位序列号>

<?php

class Snowflake
{
    // 开始时间戳（2024-01-01 00:00:00），可自定义
    const EPOCH = 1704067200000;

    // 位数分配（可根据业务调整）
    const WORKER_ID_BITS = 10;
    const SEQUENCE_BITS = 12;

    const MAX_WORKER_ID = (1 << self::WORKER_ID_BITS) - 1; // 1023
    const MAX_SEQUENCE = (1 << self::SEQUENCE_BITS) - 1;    // 4095

    // 偏移量
    const WORKER_ID_SHIFT = self::SEQUENCE_BITS;
    const TIMESTAMP_SHIFT = self::SEQUENCE_BITS + self::WORKER_ID_BITS;

    private int $workerId;
    private int $sequence = 0;
    private int $lastTimestamp = -1;
    private int $maxClockBackwardsMs = 2000; // 最大容忍时钟回拨毫秒数

    /**
     * @param int $workerId 机器ID（0~1023）
     */
    public function __construct(int $workerId)
    {
        if ($workerId < 0 || $workerId > self::MAX_WORKER_ID) {
            throw new \InvalidArgumentException("Worker ID must be between 0 and " . self::MAX_WORKER_ID);
        }
        $this->workerId = $workerId;
    }

    /**
     * 生成下一个ID
     * @return int 64位长整型ID
     */
    public function nextId(): int
    {
        $timestamp = $this->currentTimeMillis();

        // 处理时钟回拨
        if ($timestamp < $this->lastTimestamp) {
            $drift = $this->lastTimestamp - $timestamp;
            if ($drift > $this->maxClockBackwardsMs) {
                throw new \RuntimeException(
                    "Clock moved backwards too much. Refusing to generate id for {$drift}ms"
                );
            }

            // 短期回拨：沿用上一次时间戳，继续累加序列号
            $timestamp = $this->lastTimestamp;
        }

        if ($timestamp === $this->lastTimestamp) {
            // 同一毫秒内：序列号自增
            $this->sequence = ($this->sequence + 1) & self::MAX_SEQUENCE;
            if ($this->sequence === 0) {
                // 序列号用尽：必须等到下一毫秒
                $timestamp = $this->waitNextMillisecond(
                    $this->lastTimestamp === $timestamp ? $this->lastTimestamp : $timestamp
                );
            }
        } else {
            // 时间前进：序列号归零
            $this->sequence = 0;
        }

        $this->lastTimestamp = $timestamp;

        return (($timestamp - self::EPOCH) << self::TIMESTAMP_SHIFT)
               | ($this->workerId << self::WORKER_ID_SHIFT)
               | $this->sequence;
    }

    /**
     * 获取当前毫秒时间戳
     */
    private function currentTimeMillis(): int
    {
        return (int)(microtime(true) * 1000);
    }

    /**
     * 阻塞到下一毫秒
     */
    private function waitNextMillisecond(int $lastTimestamp): int
    {
        $timestamp = $this->currentTimeMillis();
        while ($timestamp <= $lastTimestamp) {
            $timestamp = $this->currentTimeMillis();
            // 可适当加入 usleep 避免 CPU 空转
            usleep(100);
        }
        return $timestamp;
    }

    /**
     * 从ID中解析出时间戳（用于调试）
     */
    public static function extractTimestamp(int $id): int
    {
        return ($id >> self::TIMESTAMP_SHIFT) + self::EPOCH;
    }
}

CREATE TABLE `id_segment` (
  `biz_tag` varchar(64) NOT NULL COMMENT '业务标识',
  `max_id` bigint(20) NOT NULL DEFAULT '0' COMMENT '当前已分配的最大ID',
  `step` int(11) NOT NULL DEFAULT '2000' COMMENT '每次分配的号段长度',
  `update_time` datetime NOT NULL,
  PRIMARY KEY (`biz_tag`)
) ENGINE=InnoDB;

UPDATE id_segment SET max_id = max_id + step WHERE biz_tag = 'order';
-- 事务提交后，本次分配范围：(old_max_id + 1) ~ (old_max_id + step)

<?php

class SegmentIdGenerator
{
    private PDO $pdo;
    private string $bizTag;
    private int $step;
    private int $currentId;
    private int $maxId;
    private int $bufferThreshold; // 0~1，触发异步加载的阈值

    public function __construct(PDO $pdo, string $bizTag, int $step = 2000, float $bufferThreshold = 0.9)
    {
        $this->pdo = $pdo;
        $this->bizTag = $bizTag;
        $this->step = $step;
        $this->bufferThreshold = $bufferThreshold;
        // 初始化时加载第一个号段
        $this->loadNextSegment();
    }

    /**
     * 从数据库获取下一个号段
     */
    private function loadNextSegment(): void
    {
        $this->pdo->beginTransaction();
        try {
            // 行级锁获取当前max_id并更新
            $stmt = $this->pdo->prepare("SELECT max_id FROM id_segment WHERE biz_tag = ? FOR UPDATE");
            $stmt->execute([$this->bizTag]);
            $row = $stmt->fetch(PDO::FETCH_ASSOC);
            if (!$row) {
                // 如果业务标志不存在，需要先初始化（示例简化，实际应插入并重试）
                $this->pdo->prepare("INSERT INTO id_segment (biz_tag, max_id, step, update_time) VALUES (?, 0, ?, NOW())")
                    ->execute([$this->bizTag, $this->step]);
                $currentMax = 0;
            } else {
                $currentMax = (int)$row['max_id'];
            }

            $newMax = $currentMax + $this->step;
            $this->pdo->prepare("UPDATE id_segment SET max_id = ?, update_time = NOW() WHERE biz_tag = ?")
                ->execute([$newMax, $this->bizTag]);

            $this->pdo->commit();

            // 设置内存中的号段范围：[currentId, maxId]
            $this->currentId = $currentMax + 1;
            $this->maxId = $newMax;
        } catch (\Exception $e) {
            $this->pdo->rollBack();
            throw new \RuntimeException("Load segment failed: " . $e->getMessage());
        }
    }

    /**
     * 获取一个ID
     */
    public function nextId(): int
    {
        // 如果当前号段已用完，加载下一段
        if ($this->currentId > $this->maxId) {
            $this->loadNextSegment();
        }

        $id = $this->currentId;
        $this->currentId++;

        // 异步预加载：当前段消费超过阈值时，触发后台拉取下一段
        $consumed = $this->currentId - ($this->maxId - $this->step + 1);
        if ($consumed >= $this->step * $this->bufferThreshold) {
            // 实际项目应放入异步任务/协程，这里简化为同步
            // 可使用Swoole\Coroutine::create 或 ReactPHP等
        }

        return $id;
    }
}

需要严格递增吗？
 ├─ 是 → 暴露业务量敏感吗？
 │       ├─ 不敏感 → 号段模式（双Buffer + DB）
 │       └─ 敏感 → 号段模式 + ID加解密混淆
 └─ 否（趋势递增即可） → 对时钟回拨零容忍？
         ├─ 否 → 雪花算法（PHP实现带容错）
         └─ 是 → 号段模式或基于Redis的原子自增