Twitter の分散型自己増加 ID アルゴリズム スノーフレーク

Twitter-Snowflake アルゴリズムの背景は非常に単純です。1 秒あたり数万件のメッセージに対する Twitter の要求を満たすには、各メッセージに一意の ID を割り当てる必要があります。これらの ID は、(処理を容易にするために) ある程度順序付けされている必要があります。分散システムでは、異なるマシンによって生成される ID は異なる必要があります。

Snowflake アルゴリズム コア

は、タイムスタンプ、作業マシン ID、シリアル番号を組み合わせます。

使用不可としてマークされた最上位ビットを除き、特定のビジネス ニーズに応じて、ビット位置の残りの 3 つのグループをフローティングにすることができます。デフォルトでは、41 ビットのタイムスタンプは 2082 年までこのアルゴリズムの使用をサポートし、10 ビットの作業マシン ID は 1023 台のマシンをサポートし、シリアル番号は 4095 個の自動インクリメント シーケンス ID の生成に 1 ミリ秒をサポートします。これについては以下で詳しく分析します。

Snowflake – タイムスタンプ

ここでのタイムスタンプの粒度はミリ秒レベルです。vdso のため、gettimeofday() は 64 ビット Linux システムマシンを使用することをお勧めします。ユーザー モードでの操作により、カーネル状態に入る際の損失が軽減されます。

uint64_t generateStamp()
{
    timeval tv;
    gettimeofday(&tv, 0);
    return (uint64_t)tv.tv_sec * 1000 + (uint64_t)tv.tv_usec / 1000;
}

デフォルトでは、使用可能なビットが 41 ビットあるため、使用および割り当てできる合計 T (1llu << 41) ミリ秒があります。年 = T / (3600 * 24 * 365 * 1000) = 69.7 年。タイムスタンプに 39 ビットのみを割り当てた場合、同じアルゴリズムによると、最終年は 17.4 年になります。

Snowflake – 作業マシン ID

厳密に言えば、このビット セグメントはプロセス レベルで使用でき、作業プロセス レベルで MAC アドレスを使用して作業マシンを一意に識別できます。 IP+パスを使用して作業プロセスを区別できます。稼働中のマシンが比較的少ない場合は、構成ファイルを使用してこの ID を設定することをお勧めします。マシンが多すぎる場合は、構成ファイルを維持するのが大変なことになります。

ここでの解決策は、作業 ID を割り当てるプロセスが必要であるということです。割り当て ID を記録する簡単なプロセスを自分で作成することも、Mysql auto_increment メカニズムを使用して効果を実現することもできます。

ワークプロセスとワークIDアロケーターは、ワークプロセスの開始時に一度だけ対話します。その後、ワークプロセスは割り当てられたIDデータをそれ自体でファイルに入れることができ、次回の開始時に、ワークプロセスが直接読み取ります。使用するファイル内の ID。

追記: この動作中のマシン ID のビット セグメントはさらに分割することもできます。たとえば、最初の 5 ビットはプロセス ID をマークするために使用され、最後の 5 ビットはスレッド ID をマークするために使用されます: D

Snowflake。 – シリアル番号

シリアル番号は 1 つのシリーズの自動インクリメント ID (マルチスレッドにはアトミックが推奨されます)。シーケンス番号が割り当てられている場合は、複数のメッセージを処理する必要があります。同じミリ秒内に使い切る場合は、「次のミリ秒まで待ちます」。

uint64_t waitNextMs(uint64_t lastStamp)
{
    uint64_t cur = 0;
    do {
        cur = generateStamp();
    } while (cur <= lastStamp);
    return cur;
}

一般的に言えば、現在の主流の 128 ビット GUID アルゴリズムとは異なり、厳密な ID シリアル化が保証されない場合でも、int64_t フィールドで機能します。ゲームサーバー側でのGUID生成として使用すると非常に便利です。さらに、マルチスレッド環境では、シーケンス番号にアトミックを使用すると、コード実装のロック密度を効果的に削減できます。

分散システムでは、グローバル UID を生成する必要がある場合が多くありますが、Twitter のスノーフレークはこのニーズを解決します。構成情報を除けば、コア コードは 41 ビットのミリ秒時間 + マシン ID です。 10 ビット + ミリ秒で 12 ビットをシーケンスします。

コア コードは IdWorker クラスによって実装されており、その原理構造は次のとおりです。ビットを表すために 0 を使用し、各部分の機能を区切るために - を使用します。

0---0000000000 0000000000 0000000000 0000000000 0 --- 00000 ---00000 ---000000000000

在上面的字符串中，第一位为未使用（实际上也可作为long的符号位），接下来的41位为毫秒级时间，然后5位datacenter标识位，5位机器ID（并不算标识符，实际是为线程标识），然后12位该毫秒内的当前毫秒内的计数，加起来刚好64位，为一个Long型。

这样的好处是，整体上按照时间自增排序，并且整个分布式系统内不会产生ID碰撞（由datacenter和机器ID作区分），并且效率较高，经测试，snowflake每秒能够产生26万ID左右，完全满足需要。

且看其核心代码：

/** Copyright 2010-2012 Twitter, Inc.*/
package com.twitter.service.snowflake

import com.twitter.ostrich.stats.Stats
import com.twitter.service.snowflake.gen._
import java.util.Random
import com.twitter.logging.Logger

/**
 * An object that generates IDs.
 * This is broken into a separate class in case
 * we ever want to support multiple worker threads
 * per process
 */
class IdWorker(val workerId: Long, val datacenterId: Long, private val reporter: Reporter, var sequence: Long = 0L)
extends Snowflake.Iface {
  private[this] def genCounter(agent: String) = {
    Stats.incr("ids_generated")
    Stats.incr("ids_generated_%s".format(agent))
  }
  private[this] val exceptionCounter = Stats.getCounter("exceptions")
  private[this] val log = Logger.get
  private[this] val rand = new Random

  val twepoch = 1288834974657L

 //机器标识位数

  private[this] val workerIdBits = 5L

//数据中心标识位数
  private[this] val datacenterIdBits = 5L

//机器ID最大值
  private[this] val maxWorkerId = -1L ^ (-1L << workerIdBits)

//数据中心ID最大值
  private[this] val maxDatacenterId = -1L ^ (-1L << datacenterIdBits)

//毫秒内自增位
  private[this] val sequenceBits = 12L

//机器ID偏左移12位

  private[this] val workerIdShift = sequenceBits

//数据中心ID左移17位
  private[this] val datacenterIdShift = sequenceBits + workerIdBits

//时间毫秒左移22位
  private[this] val timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits
  private[this] val sequenceMask = -1L ^ (-1L << sequenceBits)

  private[this] var lastTimestamp = -1L

  // sanity check for workerId
  if (workerId > maxWorkerId || workerId < 0) {
    exceptionCounter.incr(1)
    throw new IllegalArgumentException("worker Id can&#39;t be greater than %d or less than 0".format(maxWorkerId))
  }

  if (datacenterId > maxDatacenterId || datacenterId < 0) {
    exceptionCounter.incr(1)
    throw new IllegalArgumentException("datacenter Id can&#39;t be greater than %d or less than 0".format(maxDatacenterId))
  }

  log.info("worker starting. timestamp left shift %d, datacenter id bits %d, worker id bits %d, sequence bits %d, workerid %d",
    timestampLeftShift, datacenterIdBits, workerIdBits, sequenceBits, workerId)

  def get_id(useragent: String): Long = {
    if (!validUseragent(useragent)) {
      exceptionCounter.incr(1)
      throw new InvalidUserAgentError
    }

    val id = nextId()
    genCounter(useragent)

    reporter.report(new AuditLogEntry(id, useragent, rand.nextLong))
    id
  }

  def get_worker_id(): Long = workerId
  def get_datacenter_id(): Long = datacenterId
  def get_timestamp() = System.currentTimeMillis

  protected[snowflake] def nextId(): Long = synchronized {
    var timestamp = timeGen()

 //时间错误

    if (timestamp < lastTimestamp) {
      exceptionCounter.incr(1)
      log.error("clock is moving backwards.  Rejecting requests until %d.", lastTimestamp);
      throw new InvalidSystemClock("Clock moved backwards.  Refusing to generate id for %d milliseconds".format(
        lastTimestamp - timestamp))
    }

    if (lastTimestamp == timestamp) {
//当前毫秒内，则+1
      sequence = (sequence + 1) & sequenceMask
      if (sequence == 0) {
//当前毫秒内计数满了，则等待下一秒
        timestamp = tilNextMillis(lastTimestamp)
      }
    } else {
      sequence = 0
    }

    lastTimestamp = timestamp
//ID偏移组合生成最终的ID，并返回ID   

((timestamp - twepoch) << timestampLeftShift) |
      (datacenterId << datacenterIdShift) |
      (workerId << workerIdShift) |
      sequence
  }

//等待下一个毫秒的到来 

protected def tilNextMillis(lastTimestamp: Long): Long = {
    var timestamp = timeGen()
    while (timestamp <= lastTimestamp) {
      timestamp = timeGen()
    }
    timestamp
  }

  protected def timeGen(): Long = System.currentTimeMillis()

  val AgentParser = """([a-zA-Z][a-zA-Z\-0-9]*)""".r

  def validUseragent(useragent: String): Boolean = useragent match {
    case AgentParser(_) => true
    case _ => false
  }
}

上述为twitter的实现，下面且看一个Java实现，貌似为淘宝的朋友写的。

public class IdWorker {
 private final long workerId;
 private final static long twepoch = 1361753741828L;
 private long sequence = 0L;
 private final static long workerIdBits = 4L;
 public final static long maxWorkerId = -1L ^ -1L << workerIdBits;
 private final static long sequenceBits = 10L;
 private final static long workerIdShift = sequenceBits;
 private final static long timestampLeftShift = sequenceBits + workerIdBits;
 public final static long sequenceMask = -1L ^ -1L << sequenceBits;
 private long lastTimestamp = -1L;
 public IdWorker(final long workerId) {
  super();
  if (workerId > this.maxWorkerId || workerId < 0) {
   throw new IllegalArgumentException(String.format(
     "worker Id can&#39;t be greater than %d or less than 0",
     this.maxWorkerId));
  }
  this.workerId = workerId;
 }
 public synchronized long nextId() {
  long timestamp = this.timeGen();
  if (this.lastTimestamp == timestamp) {
   this.sequence = (this.sequence + 1) & this.sequenceMask;
   if (this.sequence == 0) {
    System.out.println("###########" + sequenceMask);
    timestamp = this.tilNextMillis(this.lastTimestamp);
   }
  } else {
   this.sequence = 0;
  }
  if (timestamp < this.lastTimestamp) {
   try {
    throw new Exception(
      String.format(
        "Clock moved backwards.  Refusing to generate id for %d milliseconds",
        this.lastTimestamp - timestamp));
   } catch (Exception e) {
    e.printStackTrace();
   }
  }
  this.lastTimestamp = timestamp;
  long nextId = ((timestamp - twepoch << timestampLeftShift))
    | (this.workerId << this.workerIdShift) | (this.sequence);
//  System.out.println("timestamp:" + timestamp + ",timestampLeftShift:"
//    + timestampLeftShift + ",nextId:" + nextId + ",workerId:"
//    + workerId + ",sequence:" + sequence);
  return nextId;
 }
 private long tilNextMillis(final long lastTimestamp) {
  long timestamp = this.timeGen();
  while (timestamp <= lastTimestamp) {
   timestamp = this.timeGen();
  }
  return timestamp;
 }
 private long timeGen() {
  return System.currentTimeMillis();
 }
 
 
 public static void main(String[] args){
  IdWorker worker2 = new IdWorker(2);
  System.out.println(worker2.nextId());
  
 }
}

再来看一个php的实现

<?php
class Idwork
{
const debug = 1;
static $workerId;
static $twepoch = 1361775855078;
static $sequence = 0;
const workerIdBits = 4;
static $maxWorkerId = 15;
const sequenceBits = 10;
static $workerIdShift = 10;
static $timestampLeftShift = 14;
static $sequenceMask = 1023;
private  static $lastTimestamp = -1;
function __construct($workId){
if($workId > self::$maxWorkerId || $workId< 0 )
{
throw new Exception("worker Id can&#39;t be greater than 15 or less than 0");
}
self::$workerId=$workId;
echo &#39;logdebug->__construct()->self::$workerId:&#39;.self::$workerId;
echo &#39;</br>&#39;;
}
function timeGen(){
//获得当前时间戳
$time = explode(&#39; &#39;, microtime());
$time2= substr($time[0], 2, 3);
$timestramp = $time[1].$time2;
echo &#39;logdebug->timeGen()->$timestramp:&#39;.$time[1].$time2;
echo &#39;</br>&#39;;
return  $time[1].$time2;
}
function  tilNextMillis($lastTimestamp) {
$timestamp = $this->timeGen();
while ($timestamp <= $lastTimestamp) {
$timestamp = $this->timeGen();
}
echo &#39;logdebug->tilNextMillis()->$timestamp:&#39;.$timestamp;
echo &#39;</br>&#39;;
return $timestamp;
}
function  nextId()
{
$timestamp=$this->timeGen();
echo &#39;logdebug->nextId()->self::$lastTimestamp1:&#39;.self::$lastTimestamp;
echo &#39;</br>&#39;;
if(self::$lastTimestamp == $timestamp) {
self::$sequence = (self::$sequence + 1) & self::$sequenceMask;
if (self::$sequence == 0) {
    echo "###########".self::$sequenceMask;
    $timestamp = $this->tilNextMillis(self::$lastTimestamp);
    echo &#39;logdebug->nextId()->self::$lastTimestamp2:&#39;.self::$lastTimestamp;
    echo &#39;</br>&#39;;
  }
} else {
self::$sequence  = 0;
    echo &#39;logdebug->nextId()->self::$sequence:&#39;.self::$sequence;
    echo &#39;</br>&#39;;
}
if ($timestamp < self::$lastTimestamp) {
   throw new Excwption("Clock moved backwards.  Refusing to generate id for ".(self::$lastTimestamp-$timestamp)." milliseconds");
   }
self::$lastTimestamp  = $timestamp;
echo &#39;logdebug->nextId()->self::$lastTimestamp3:&#39;.self::$lastTimestamp;
echo &#39;</br>&#39;;
echo &#39;logdebug->nextId()->(($timestamp - self::$twepoch << self::$timestampLeftShift )):&#39;.((sprintf(&#39;%.0f&#39;, $timestamp) - sprintf(&#39;%.0f&#39;, self::$twepoch) ));
echo &#39;</br>&#39;;
$nextId = ((sprintf(&#39;%.0f&#39;, $timestamp) - sprintf(&#39;%.0f&#39;, self::$twepoch)  )) | ( self::$workerId << self::$workerIdShift ) | self::$sequence;
echo &#39;timestamp:&#39;.$timestamp.&#39;-----&#39;;
echo &#39;twepoch:&#39;.sprintf(&#39;%.0f&#39;, self::$twepoch).&#39;-----&#39;;
echo &#39;timestampLeftShift =&#39;.self::$timestampLeftShift.&#39;-----&#39;;
echo &#39;nextId:&#39;.$nextId.&#39;----&#39;;
echo &#39;workId:&#39;.self::$workerId.&#39;-----&#39;;
echo &#39;workerIdShift:&#39;.self::$workerIdShift.&#39;-----&#39;;
return $nextId;
}
}
$Idwork = new Idwork(1);
$a= $Idwork->nextId();
$Idwork = new Idwork(2);
$a= $Idwork->nextId();
?>