【系统架构设计百科】Twitter 架构演进：从 Ruby 到分布式时间线

# 早期 Twitter 的时间线查询（简化示意）
class TimelineController < ApplicationController
  def home
    following_ids = current_user.following.pluck(:id)
    @tweets = Tweet.where(user_id: following_ids)
                   .order(created_at: :desc)
                   .limit(200)
  end
end

// Finagle Thrift 服务端示例
import com.twitter.finagle.Thrift
import com.twitter.util.{Await, Future}

object TweetServiceServer {
  def main(args: Array[String]): Unit = {
    val service = new TweetService.MethodPerEndpoint {
      override def getTweet(tweetId: Long): Future[Tweet] = {
        // 从 Manhattan 读取推文
        manhattanClient.get(s"tweets/$tweetId").map { result =>
          Tweet(
            id = tweetId,
            text = result.getString("text"),
            userId = result.getLong("user_id"),
            createdAt = result.getLong("created_at")
          )
        }
      }

      override def postTweet(userId: Long, text: String): Future[Tweet] = {
        val tweetId = snowflakeIdGenerator.nextId()
        val tweet = Tweet(tweetId, text, userId, System.currentTimeMillis())
        for {
          _ <- manhattanClient.put(s"tweets/$tweetId", tweet.toBytes)
          _ <- fanoutService.fanout(tweet)
        } yield tweet
      }
    }

    val server = Thrift.server
      .withLabel("tweet-service")
      .withRequestTimeout(500.milliseconds)
      .serveIface("0.0.0.0:9090", service)

    Await.ready(server)
  }
}

// Finagle 客户端示例，内置重试与熔断
import com.twitter.finagle.Thrift
import com.twitter.conversions.DurationOps._

val client = Thrift.client
  .withLabel("tweet-client")
  .withRequestTimeout(200.milliseconds)
  .withRetryBudget(RetryBudget(
    ttl = 10.seconds,
    minRetriesPerSec = 5,
    percentCanRetry = 0.1
  ))
  .build[TweetService.MethodPerEndpoint]("zk!myzkhost:2181!/twitter/services/tweet")

// 双写与校验伪代码
public class DualWriteService {
    private final MysqlDao mysqlDao;
    private final ManhattanClient manhattanClient;
    private final ConsistencyChecker checker;

    public void writeTweet(Tweet tweet) {
        // 主写入：MySQL（当前主存储）
        mysqlDao.insert(tweet);

        // 影子写入：Manhattan（新存储）
        try {
            manhattanClient.put(tweet.getId(), tweet.serialize());
        } catch (Exception e) {
            // 影子写入失败不影响主流程
            metrics.increment("shadow_write_failure");
            repairQueue.enqueue(tweet.getId());
        }

        // 异步校验
        checker.scheduleVerification(tweet.getId(), Duration.ofMinutes(5));
    }
}

# 时间线缓存结构示意（Python 伪代码）
import redis

class TimelineCache:
    def __init__(self, redis_cluster):
        self.redis = redis_cluster
        self.MAX_TIMELINE_SIZE = 800  # 每个用户最多缓存 800 条推文 ID

    def add_tweet(self, user_id: int, tweet_id: int) -> None:
        """将推文 ID 添加到用户时间线"""
        key = f"timeline:{user_id}"
        pipeline = self.redis.pipeline()
        # 使用 Snowflake ID 作为分数（包含时间信息）
        pipeline.zadd(key, {str(tweet_id): float(tweet_id)})
        # 裁剪，只保留最新的 800 条
        pipeline.zremrangebyrank(key, 0, -(self.MAX_TIMELINE_SIZE + 1))
        pipeline.execute()

    def get_timeline(self, user_id: int, count: int = 200,
                     max_id: int = None) -> list:
        """读取用户时间线"""
        key = f"timeline:{user_id}"
        if max_id:
            # 分页：获取 max_id 之前的推文
            return self.redis.zrevrangebyscore(
                key, max_id - 1, "-inf", start=0, num=count
            )
        return self.redis.zrevrange(key, 0, count - 1)

    def remove_tweet(self, user_id: int, tweet_id: int) -> None:
        """用户删除推文时，从所有粉丝时间线移除"""
        key = f"timeline:{user_id}"
        self.redis.zrem(key, str(tweet_id))

// Snowflake ID 生成器核心逻辑
public class SnowflakeIdGenerator {
    // 起始时间戳：2010-11-04T01:42:54.657Z
    private static final long EPOCH = 1288834974657L;
    private static final int WORKER_ID_BITS = 10;
    private static final int SEQUENCE_BITS = 12;
    private static final long MAX_SEQUENCE = (1L << SEQUENCE_BITS) - 1;

    private final long workerId;
    private long sequence = 0L;
    private long lastTimestamp = -1L;

    public SnowflakeIdGenerator(long workerId) {
        if (workerId < 0 || workerId >= (1L << WORKER_ID_BITS)) {
            throw new IllegalArgumentException(
                "Worker ID must be between 0 and " + ((1L << WORKER_ID_BITS) - 1)
            );
        }
        this.workerId = workerId;
    }

    public synchronized long nextId() {
        long timestamp = System.currentTimeMillis();

        if (timestamp < lastTimestamp) {
            throw new RuntimeException(
                "Clock moved backwards. Refusing to generate id for "
                + (lastTimestamp - timestamp) + " milliseconds"
            );
        }

        if (timestamp == lastTimestamp) {
            sequence = (sequence + 1) & MAX_SEQUENCE;
            if (sequence == 0) {
                // 当前毫秒的序列号用完，等待下一毫秒
                timestamp = waitNextMillis(lastTimestamp);
            }
        } else {
            sequence = 0;
        }

        lastTimestamp = timestamp;

        return ((timestamp - EPOCH) << (WORKER_ID_BITS + SEQUENCE_BITS))
             | (workerId << SEQUENCE_BITS)
             | sequence;
    }

    private long waitNextMillis(long lastTimestamp) {
        long timestamp = System.currentTimeMillis();
        while (timestamp <= lastTimestamp) {
            timestamp = System.currentTimeMillis();
        }
        return timestamp;
    }
}

// 混合扇出策略的时间线读取（Scala 伪代码）
class HybridTimelineService(
  timelineCache: TimelineCache,
  tweetStore: TweetStore,
  socialGraph: SocialGraphService,
  celebrityThreshold: Int = 100000
) {

  def getHomeTimeline(userId: Long, count: Int = 200): Future[Seq[Tweet]] = {
    // 并行执行两个查询
    val cachedTweetIdsFuture = timelineCache.get(userId, count * 2)
    val celebrityFollowingsFuture = socialGraph.getCelebrityFollowings(
      userId, celebrityThreshold
    )

    for {
      cachedTweetIds <- cachedTweetIdsFuture
      celebrityIds <- celebrityFollowingsFuture

      // 查询名人的最新推文
      celebrityTweets <- Future.collect(
        celebrityIds.map { celebId =>
          tweetStore.getLatestTweets(celebId, count = 10)
        }
      ).map(_.flatten)

      // 合并并去重
      allTweetIds = mergeTweetIds(cachedTweetIds, celebrityTweets.map(_.id))
                      .distinct
                      .sorted(Ordering[Long].reverse)
                      .take(count)

      // 水合推文（获取完整信息）
      hydratedTweets <- tweetHydrator.hydrate(allTweetIds)
    } yield hydratedTweets
  }

  private def mergeTweetIds(
    cached: Seq[Long], celebrity: Seq[Long]
  ): Seq[Long] = {
    // 基于 Snowflake ID 的时间戳进行归并排序
    val merged = new ArrayBuffer[Long](cached.length + celebrity.length)
    var i = 0
    var j = 0
    while (i < cached.length && j < celebrity.length) {
      if (cached(i) >= celebrity(j)) {
        merged += cached(i)
        i += 1
      } else {
        merged += celebrity(j)
        j += 1
      }
    }
    while (i < cached.length) { merged += cached(i); i += 1 }
    while (j < celebrity.length) { merged += celebrity(j); j += 1 }
    merged.toSeq
  }
}

// Fanout Service 核心逻辑
public class FanoutService {
    private static final int CELEBRITY_THRESHOLD = 100_000;
    private static final int BATCH_SIZE = 1000;

    private final SocialGraphClient socialGraph;
    private final TimelineCacheClient timelineCache;
    private final ExecutorService executor;
    private final Metrics metrics;

    public void fanout(Tweet tweet) {
        long userId = tweet.getUserId();
        int followerCount = socialGraph.getFollowerCount(userId);

        if (followerCount >= CELEBRITY_THRESHOLD) {
            // 名人用户：不扇出，记录指标
            metrics.increment("fanout.celebrity.skipped");
            return;
        }

        // 普通用户：分批扇出
        metrics.increment("fanout.normal.started");
        long startTime = System.currentTimeMillis();

        Cursor cursor = socialGraph.getFollowersCursor(userId);
        List<Future<?>> futures = new ArrayList<>();

        while (cursor.hasNext()) {
            List<Long> batch = cursor.next(BATCH_SIZE);
            futures.add(executor.submit(() -> {
                for (long followerId : batch) {
                    try {
                        timelineCache.addTweet(followerId, tweet.getId());
                    } catch (Exception e) {
                        metrics.increment("fanout.write.failure");
                        // 写入失败的条目进入重试队列
                        retryQueue.enqueue(followerId, tweet.getId());
                    }
                }
            }));
        }

        // 等待所有批次完成
        for (Future<?> future : futures) {
            try {
                future.get(30, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                metrics.increment("fanout.batch.timeout");
            }
        }

        long elapsed = System.currentTimeMillis() - startTime;
        metrics.recordTime("fanout.duration", elapsed);
        metrics.increment("fanout.normal.completed");
    }
}

# Manhattan 数据模型示例
dataset: tweets
  key: tweet_id (Long)
  columns:
    - name: text
      type: String
    - name: user_id
      type: Long
    - name: created_at
      type: Long
    - name: retweet_count
      type: Counter
    - name: favorite_count
      type: Counter
    - name: media_ids
      type: List<Long>

dataset: user_tweets
  key: [user_id, tweet_id] (Composite Key)
  columns:
    - name: created_at
      type: Long

dataset: timelines
  key: user_id (Long)
  columns:
    - name: tweet_ids
      type: SortedSet<Long>

// Manhattan 客户端使用示例
public class TweetRepository {
    private final ManhattanClient manhattan;
    private static final String DATASET = "tweets";

    public Optional<Tweet> getTweet(long tweetId) {
        ManhattanKey key = ManhattanKey.of(DATASET, Bytes.fromLong(tweetId));
        ManhattanResult result = manhattan.get(key, ConsistencyLevel.ONE);

        if (result.isEmpty()) {
            return Optional.empty();
        }

        return Optional.of(Tweet.builder()
            .id(tweetId)
            .text(result.getString("text"))
            .userId(result.getLong("user_id"))
            .createdAt(result.getLong("created_at"))
            .retweetCount(result.getCounter("retweet_count"))
            .favoriteCount(result.getCounter("favorite_count"))
            .build());
    }

    public void saveTweet(Tweet tweet) {
        ManhattanKey key = ManhattanKey.of(DATASET, Bytes.fromLong(tweet.getId()));

        ManhattanMutation mutation = ManhattanMutation.builder()
            .put("text", tweet.getText())
            .put("user_id", tweet.getUserId())
            .put("created_at", tweet.getCreatedAt())
            .putCounter("retweet_count", 0)
            .putCounter("favorite_count", 0)
            .build();

        manhattan.mutate(key, mutation, ConsistencyLevel.QUORUM);
    }

    public void incrementRetweetCount(long tweetId) {
        ManhattanKey key = ManhattanKey.of(DATASET, Bytes.fromLong(tweetId));
        manhattan.incrementCounter(key, "retweet_count", 1);
    }
}

// 布隆过滤器在 SSTable 查找中的应用
public class SSTableReader {
    private final BloomFilter bloomFilter;
    private final IndexBlock indexBlock;
    private final RandomAccessFile dataFile;

    public Optional<byte[]> get(byte[] key) {
        // 先检查布隆过滤器，O(1) 时间复杂度
        if (!bloomFilter.mightContain(key)) {
            // 一定不存在，避免磁盘 IO
            return Optional.empty();
        }

        // 布隆过滤器返回 true，可能存在
        // 在索引块中二分查找
        long offset = indexBlock.findOffset(key);
        if (offset < 0) {
            return Optional.empty();
        }

        // 读取数据块
        byte[] value = readDataBlock(offset, key);
        return Optional.ofNullable(value);
    }
}

// 暗读取（Dark Read）实现
public class DarkReadService {
    private final MysqlDao mysql;
    private final ManhattanClient manhattan;
    private final double darkReadPercentage;

    public Tweet getTweet(long tweetId) {
        // 始终从 MySQL 读取（主路径）
        Tweet mysqlResult = mysql.getTweet(tweetId);

        // 按比例执行暗读取
        if (ThreadLocalRandom.current().nextDouble() < darkReadPercentage) {
            CompletableFuture.runAsync(() -> {
                try {
                    Tweet manhattanResult = manhattan.getTweet(tweetId);
                    if (!mysqlResult.equals(manhattanResult)) {
                        metrics.increment("dark_read.mismatch");
                        alertService.log(
                            "Data mismatch for tweet " + tweetId
                        );
                    } else {
                        metrics.increment("dark_read.match");
                    }
                } catch (Exception e) {
                    metrics.increment("dark_read.error");
                }
            });
        }

        return mysqlResult;
    }
}

# 数据校验管道（Python 伪代码）
class DataVerificationPipeline:
    def __init__(self, mysql_client, manhattan_client, sample_rate=0.01):
        self.mysql = mysql_client
        self.manhattan = manhattan_client
        self.sample_rate = sample_rate
        self.mismatch_count = 0
        self.total_checked = 0

    def verify_batch(self, tweet_ids: list) -> dict:
        """批量校验推文数据一致性"""
        results = {
            "matched": 0,
            "mismatched": 0,
            "mysql_only": 0,
            "manhattan_only": 0,
            "errors": 0
        }

        for tweet_id in tweet_ids:
            try:
                mysql_tweet = self.mysql.get_tweet(tweet_id)
                manhattan_tweet = self.manhattan.get_tweet(tweet_id)

                if mysql_tweet is None and manhattan_tweet is None:
                    continue
                elif mysql_tweet is None:
                    results["manhattan_only"] += 1
                    self.repair_to_mysql(tweet_id, manhattan_tweet)
                elif manhattan_tweet is None:
                    results["mysql_only"] += 1
                    self.repair_to_manhattan(tweet_id, mysql_tweet)
                elif self.compare_tweets(mysql_tweet, manhattan_tweet):
                    results["matched"] += 1
                else:
                    results["mismatched"] += 1
                    self.resolve_conflict(tweet_id, mysql_tweet, manhattan_tweet)

            except Exception as e:
                results["errors"] += 1
                logging.error(f"Verification error for tweet {tweet_id}: {e}")

        self.total_checked += len(tweet_ids)
        return results

    def compare_tweets(self, mysql_tweet, manhattan_tweet) -> bool:
        """比较两个存储中的推文数据"""
        fields_to_compare = ["text", "user_id", "created_at"]
        return all(
            getattr(mysql_tweet, f) == getattr(manhattan_tweet, f)
            for f in fields_to_compare
        )

    def resolve_conflict(self, tweet_id, mysql_tweet, manhattan_tweet):
        """冲突解决：以 MySQL（主存储）为准"""
        self.manhattan.put_tweet(tweet_id, mysql_tweet)
        logging.warning(
            f"Conflict resolved for tweet {tweet_id}, "
            f"MySQL version applied to Manhattan"
        )

// Kestrel 消息队列的使用示例
import com.twitter.finagle.kestrel.Client

object FanoutWorker {
  val kestrelClient = Client("kestrel-cluster:22133")

  def processFanoutQueue(): Unit = {
    val queue = kestrelClient.from("fanout_tasks")

    queue.foreach { message =>
      try {
        val task = FanoutTask.deserialize(message.bytes)
        performFanout(task)
        message.acknowledge()
      } catch {
        case e: Exception =>
          // 处理失败，消息将被重新投递
          message.nack()
          metrics.increment("fanout.worker.failure")
      }
    }
  }

  private def performFanout(task: FanoutTask): Unit = {
    val followers = socialGraph.getFollowers(task.userId)
    followers.grouped(1000).foreach { batch =>
      batch.foreach { followerId =>
        timelineCache.addTweet(followerId, task.tweetId)
      }
    }
  }
}

// EventBus 生产者示例
public class TweetEventProducer {
    private final EventBusProducer<TweetEvent> producer;

    public TweetEventProducer(EventBusConfig config) {
        this.producer = EventBus.newProducer(
            config,
            "tweet_events",
            TweetEvent.class
        );
    }

    public void publishTweetCreated(Tweet tweet) {
        TweetEvent event = TweetEvent.builder()
            .eventType(EventType.TWEET_CREATED)
            .tweetId(tweet.getId())
            .userId(tweet.getUserId())
            .text(tweet.getText())
            .timestamp(System.currentTimeMillis())
            .build();

        producer.publish(event).onSuccess(offset -> {
            metrics.increment("tweet.event.published");
        }).onFailure(ex -> {
            metrics.increment("tweet.event.publish_failed");
            deadLetterQueue.enqueue(event);
        });
    }
}

// EventBus 消费者示例
public class FanoutEventConsumer {
    private final EventBusConsumer<TweetEvent> consumer;
    private final FanoutService fanoutService;

    public void start() {
        consumer = EventBus.newConsumer(
            config,
            "tweet_events",
            "fanout-consumer-group",
            TweetEvent.class
        );

        consumer.subscribe(event -> {
            if (event.getEventType() == EventType.TWEET_CREATED) {
                try {
                    fanoutService.fanout(event.toTweet());
                    return ConsumerResult.SUCCESS;
                } catch (Exception e) {
                    return ConsumerResult.RETRY;
                }
            }
            return ConsumerResult.SUCCESS;
        });
    }
}

// Twemcache 客户端使用示例
public class TweetCacheService {
    private final TwemcacheClient cache;
    private static final int TWEET_TTL_SECONDS = 86400; // 24 小时
    private static final int HOT_TWEET_TTL_SECONDS = 3600; // 热点推文 1 小时

    public Optional<Tweet> getCachedTweet(long tweetId) {
        String key = "tweet:" + tweetId;
        byte[] cached = cache.get(key);
        if (cached != null) {
            metrics.increment("tweet.cache.hit");
            return Optional.of(Tweet.deserialize(cached));
        }
        metrics.increment("tweet.cache.miss");
        return Optional.empty();
    }

    public void cacheTweet(Tweet tweet) {
        String key = "tweet:" + tweet.getId();
        int ttl = isHotTweet(tweet) ? HOT_TWEET_TTL_SECONDS : TWEET_TTL_SECONDS;
        cache.set(key, tweet.serialize(), ttl);
    }

    public Map<Long, Tweet> multiGetTweets(List<Long> tweetIds) {
        // 批量获取，减少网络往返
        List<String> keys = tweetIds.stream()
            .map(id -> "tweet:" + id)
            .collect(Collectors.toList());

        Map<String, byte[]> results = cache.multiGet(keys);
        Map<Long, Tweet> tweets = new HashMap<>();

        for (int i = 0; i < tweetIds.size(); i++) {
            byte[] data = results.get(keys.get(i));
            if (data != null) {
                tweets.put(tweetIds.get(i), Tweet.deserialize(data));
            }
        }
        return tweets;
    }

    private boolean isHotTweet(Tweet tweet) {
        return tweet.getRetweetCount() > 10000
            || tweet.getFavoriteCount() > 50000;
    }
}

// 防止缓存击穿的租约机制
public class LeaseBasedCache<K, V> {
    private final TwemcacheClient cache;
    private final ConcurrentMap<K, CompletableFuture<V>> pendingLoads;

    public V getOrLoad(K key, Function<K, V> loader) {
        // 先尝试缓存
        V cached = cache.get(key);
        if (cached != null) {
            return cached;
        }

        // 尝试获取租约
        CompletableFuture<V> future = new CompletableFuture<>();
        CompletableFuture<V> existing = pendingLoads.putIfAbsent(key, future);

        if (existing != null) {
            // 已有其他线程在加载，等待结果
            try {
                return existing.get(500, TimeUnit.MILLISECONDS);
            } catch (TimeoutException e) {
                // 等待超时，直接从后端加载
                return loader.apply(key);
            }
        }

        // 获得租约，负责加载数据
        try {
            V value = loader.apply(key);
            cache.set(key, value);
            future.complete(value);
            return value;
        } catch (Exception e) {
            future.completeExceptionally(e);
            throw e;
        } finally {
            pendingLoads.remove(key);
        }
    }
}

// 混合时间线读取的降级策略
public class ResilientTimelineService {
    private final TimelineCache timelineCache;
    private final CelebrityTweetFetcher celebrityFetcher;
    private final TweetHydrator hydrator;
    private final CircuitBreaker celebrityCircuitBreaker;

    public TimelineResponse getTimeline(long userId, int count) {
        // 从缓存读取基础时间线（始终执行）
        List<Long> baseTweetIds;
        try {
            baseTweetIds = timelineCache.get(userId, count);
        } catch (Exception e) {
            // 缓存不可用时的降级：直接从存储读取
            metrics.increment("timeline.cache.fallback");
            baseTweetIds = fallbackToStorage(userId, count);
        }

        // 获取名人推文（可降级）
        List<Long> celebrityTweetIds = Collections.emptyList();
        if (celebrityCircuitBreaker.isAllowed()) {
            try {
                celebrityTweetIds = celebrityFetcher.fetch(userId, 10);
            } catch (Exception e) {
                // 名人推文获取失败：降级为只显示已缓存的推文
                metrics.increment("timeline.celebrity.degraded");
                celebrityCircuitBreaker.recordFailure();
            }
        }

        // 合并并水合
        List<Long> mergedIds = merge(baseTweetIds, celebrityTweetIds, count);

        List<Tweet> hydratedTweets;
        try {
            hydratedTweets = hydrator.hydrate(mergedIds);
        } catch (Exception e) {
            // 水合失败：返回部分结果
            metrics.increment("timeline.hydration.partial");
            hydratedTweets = hydrator.hydratePartial(mergedIds);
        }

        return new TimelineResponse(hydratedTweets);
    }

    private List<Long> fallbackToStorage(long userId, int count) {
        // 降级方案：查询用户关注列表的最新推文
        List<Long> followingIds = socialGraph.getFollowing(userId);
        return tweetStore.getLatestTweetsByUsers(followingIds, count);
    }
}

# 时间线服务的关键监控指标
monitoring:
  metrics:
    - name: timeline.read.latency.p50
      threshold: 50ms
      alert: warning
    - name: timeline.read.latency.p99
      threshold: 200ms
      alert: critical
    - name: timeline.cache.hit_rate
      threshold_below: 0.95
      alert: warning
    - name: timeline.celebrity.degraded.rate
      threshold_above: 0.05
      alert: critical
    - name: fanout.lag.seconds
      threshold_above: 30
      alert: critical
    - name: fanout.failure.rate
      threshold_above: 0.01
      alert: warning

  dashboards:
    timeline_health:
      panels:
        - title: "时间线读取延迟分布"
          type: histogram
          metrics: [timeline.read.latency]
        - title: "缓存命中率"
          type: gauge
          metrics: [timeline.cache.hit_rate]
        - title: "扇出延迟"
          type: timeseries
          metrics: [fanout.lag.seconds]
        - title: "名人推文降级率"
          type: timeseries
          metrics: [timeline.celebrity.degraded.rate]