---
title: "构建生产环境消费者"
canonical_url: "https://tweetstream.io/zh-cn/docs/production-consumer"
last_updated: 2026-09-29
---

# 构建生产环境消费者

分发已知操作，通过退避策略重连，并确保每个副作用都可安全重复。

Canonical URL: https://tweetstream.io/zh-cn/docs/production-consumer

## 通过退避策略重连

从可重连的 Node.js 或 Python 消费者开始，再用一个完整的处理程序分发所有操作。

- [可重连消费者](https://tweetstream.io/zh-cn/docs/connection-auth-heartbeat#connect): 使用指数退避的 Node.js 和 Python 客户端。
- [完整分发器](https://tweetstream.io/zh-cn/docs/production-consumer#complete-router): 覆盖全部 18 种服务端操作的类型化处理程序。

## 分发所有服务端操作

此 TypeScript 分发器覆盖全部 18 种服务端操作。它会先验证输入，并在遇到未知操作时忽略该消息，而不是关闭 WebSocket。

让每个处理程序保持简短。在调用策略、通知或执行边界前，先保存回执和本地投影。

### 完整 TypeScript 分发器

```typescript
type VerifiedType = 'blue' | 'business' | 'government' | 'verified' | 'none';
type TweetVerifiedLabel = {
  badge: string | null;
  description: string;
  url: string | null;
};

type TweetAuthor = {
  banner?: string;
  bio?: string;
  followersCount?: number;
  followingCount?: number;
  id?: string;
  joinedAt?: number;
  location?: string;
  metrics?: {
    likes?: number;
    tweets?: number;
  };
  // Includes a leading @ when present, for example "@elonmusk".
  handle?: string;
  name?: string;
  platform?: 'twitter' | 'truth_social' | 'binance_square';
  profileImage?: string;
  // The account's own website, when it has one.
  url?: string;
  verifiedLabel?: TweetVerifiedLabel;
  verifiedType?: VerifiedType;
};

type Media = {
  url: string;
} & (
  | {
      type: 'video';
      // Every video is a progressive MP4. A public still poster is included
      // when available.
      thumbnail?: string;
    }
  | {
      type?: 'image' | 'gif';
      thumbnail?: string;
    }
);

type TweetUrl = {
  url: string;
  name?: string;
  tco?: string;
};

type TweetMention = {
  handle?: string;
  id?: string;
  name?: string;
};

type TweetArticle = {
  description?: string;
  id?: string;
  publishedAt?: number;
  text?: string;
  thumbnail?: string;
  title: string;
  updatedAt?: number;
  url: string;
};

type TweetPollChoice = {
  id?: string;
  image?: string;
  label: string;
  votes?: number;
};

type TweetPoll = {
  choices: TweetPollChoice[];
  endsAt?: number;
  totalVotes?: number;
  updatedAt?: number;
};

type TweetContentKind = 'post' | 'reply' | 'quote' | 'retweet';

type TweetReference = {
  article?: TweetArticle;
  type: 'reply' | 'quote' | 'retweet';
  tweetId: string;
  text?: string;
  translatedText?: string;
  author?: TweetAuthor;
  media?: Media[];
  poll?: TweetPoll;
  quoted?: TweetReference;
  subtweet?: TweetReference;
};

type TweetContent = {
  tweetId: string;
  kind: TweetContentKind;
  // Original tweet text when the event includes both original and translated text.
  text: string;
  // Translation, present only when available.
  translatedText?: string;
  createdAt: number;
  author: TweetAuthor;
  article?: TweetArticle;
  link?: string;
  media?: Media[];
  mentions?: TweetMention[];
  poll?: TweetPoll;
  receivedAt?: number;
  urls?: TweetUrl[];
  ref?: TweetReference;
};

type TweetMeta = {
  tweetId: string;
  // Pair with tweetId when merging enrichment. When omitted, use twitter.
  platform?: 'twitter' | 'truth_social' | 'binance_square';
  ocr?: {
    text: string;
  };
  detected?: {
    tokens?: Array<{
      symbol?: string;
      name?: string;
      contract?: string;
      chain?: string;
      networkId?: number;
      priceUsd?: number;
      sources: Array<'text' | 'ocr'>;
    }>;
    cex?: Array<{
      exchange: 'bybit' | 'binance' | 'hyperliquid';
      symbol?: string;
      priceUsd?: number;
      url?: string;
      baseAsset?: string;
      quoteAsset?: string;
      sources: Array<'text' | 'ocr'>;
    }>;
    prediction?: Array<{
      exchange: 'polymarket' | 'kalshi';
      marketId?: string;
      title?: string;
      priceUsd?: number;
      url?: string;
      sources: Array<'text' | 'ocr'>;
    }>;
  };
};

type TweetUpdate = {
  tweetId: string;
  // Pair with tweetId. When absent, use author.platform, then twitter.
  platform?: 'twitter' | 'truth_social' | 'binance_square';
  article?: TweetArticle;
  kind?: TweetContentKind;
  translatedText?: string;
  author?: TweetAuthor;
  media?: Media[];
  mentions?: TweetMention[];
  poll?: TweetPoll;
  receivedAt?: number;
  urls?: TweetUrl[];
  ref?: TweetReference;
} & (
  | {
      text?: string;
      textUpdateType?: never;
    }
  | {
      text: string;
      // Supplies the full text of a previously shortened post. Not an edit.
      textUpdateType: 'completion';
    }
);

type TweetDeleteEvent = {
  tweetId: string;
  // When absent, use author.platform, then twitter.
  platform?: 'twitter' | 'truth_social' | 'binance_square';
  eventId?: string;
  deletedAt?: number;
  receivedAt?: number;
  author?: TweetAuthor;
  text?: string;
};

type TweetPinEvent = {
  tweetId: string;
  // When absent, use tweet.author.platform or author.platform, then twitter.
  platform?: 'twitter' | 'truth_social' | 'binance_square';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  action: 'pin';
  author: TweetAuthor;
  text?: string;
  tweet?: TweetContent;
};

type TweetUnpinEvent = {
  tweetId: string;
  // When absent, use tweet.author.platform or author.platform, then twitter.
  platform?: 'twitter' | 'truth_social' | 'binance_square';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  action: 'unpin';
  author: TweetAuthor;
  text?: string;
  tweet?: TweetContent;
};

type AccountEventActor = TweetAuthor & {
  websiteUrl?: string;
};

type ProfileUpdateEvent = {
  kind: 'PROFILE';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  actor: AccountEventActor;
  changes: {
    avatar?: string;
    banner?: string;
    bio?: string;
    handle?: string;
    location?: string;
    name?: string;
    verifiedLabel?: TweetVerifiedLabel | null;
    websiteUrl?: string | null;
  };
  previous?: {
    avatar?: string;
    banner?: string;
    bio?: string;
    handle?: string;
    location?: string;
    name?: string;
    verifiedLabel?: TweetVerifiedLabel | null;
    websiteUrl?: string | null;
  };
};

type FollowEvent = {
  kind: 'FOLLOW';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  actor: AccountEventActor;
  target: AccountEventActor & {
    handle: string;
  };
};

type UnfollowEvent = {
  kind: 'UNFOLLOW';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  actor: AccountEventActor;
  target: AccountEventActor & {
    handle: string;
  };
};

type AffiliateAccountIdentity = AccountEventActor & {
  id: string;
};

type AffiliateUpdateEvent = {
  action: 'added' | 'removed';
  eventId: string;
  observedAt: number;
  receivedAt?: number;
  organization: AffiliateAccountIdentity;
  member: AffiliateAccountIdentity;
};

type CalloutCaller = {
  username: string;
  image?: string;
};

type CalloutReply = {
  caller: CalloutCaller;
  replyId: string;
  parentReplyId?: string;
  rootReplyId?: string;
  text?: string;
  mediaUrl?: string;
};

type CalloutEvent = {
  platform: 'pump_fun';
  caller: CalloutCaller;
  kind?: 'callout' | 'author_update' | 'reply' | 'repost';
  text?: string;
  mediaUrl?: string;
  // Required for replies and author updates.
  replyId?: string;
  parentReplyId?: string;
  rootReplyId?: string;
  reference?: {
    caller: CalloutCaller;
    text?: string;
    mediaUrl?: string;
    reply?: CalloutReply;
  };
  position?: {
    amountHeld?: string;
    valueUsd?: string;
    costBasisUsd?: string;
    pnlUsd?: string;
    pnlPercentage?: string;
    realizedPnlUsd?: string;
  };
  token: {
    address: string;
    chainId?: string;
    symbol?: string;
    name?: string;
    image?: string;
  };
  url?: string;
  marketCapUsd?: number;
  // SOL, exact decimal.
  calloutPrice?: string;
  // USD, exact decimal.
  calloutPriceUsd?: string;
  multiple?: number;
  // SOL peak, exact decimal.
  maxPriceSol?: string;
  // USD peak, exact decimal.
  maxPriceUsd?: string;
  calloutId: string;
  createdAt: number;
  receivedAt?: number;
};

type CalloutUpdateEvent = CalloutEvent & {
  // Original envelope id, not calloutId.
  updateOf: string;
};

type FomoCoin = {
  address: string;
  networkId: number;
  name?: string;
  symbol?: string;
  image?: string;
  metadataObservedAt?: number;
  priceUsd?: string;
  marketCapUsd?: string;
  liquidityUsd?: string;
  volume24hUsd?: string;
  change24hPercent?: string;
};
type FomoThesisEvent = {
  platform: 'fomo';
  coin: FomoCoin;
  author: {
    id: string;
    username: string;
    verified: boolean;
    displayName?: string;
    avatar?: string;
    twitter?: string;
  };
  thesis: {
    id: string;
    text: string;
    createdAt: number;
    likes: number;
    replies: number;
    tradeId?: string;
    parentId?: string;
    segments?: Array<{ text: string; link?: string }>;
  };
  position?: {
    tokenAmount?: string;
    usdValue?: string;
    unrealizedPnlUsd?: string;
    realizedPnlUsd?: string;
    unrealizedPnlPercent?: string;
    realizedPnlPercent?: string;
    closedAt?: number;
  };
  observedAt?: number;
  receivedAt?: number;
};
type FomoTokenVerifiedEvent = {
  platform: 'fomo';
  coin: FomoCoin;
  observedAt: number;
  receivedAt?: number;
};

type CharityAddedEvent = {
  id: string;
  name: string;
  slug: string;
  enabled: boolean;
  taxId?: string;
  location?: {
    city?: string;
    state?: string;
    country?: string;
  };
  observedAt: number;
};

type TwitterHandlesResult = {
  action: 'follow' | 'unfollow';
  requestId: string | null;
  results: Array<{
    input: string;
    state:
      | 'added'
      | 'already_following'
      | 'invalid_input'
      | 'duplicate'
      | 'not_found'
      | 'failed'
      | 'removed'
      | 'not_following';
    message?: string;
  }>;
  error: string | null;
};

type NewsSource = {
  id: string;
  name: string;
};

type NewsMedia = {
  url: string;
  type?: 'image' | 'video';
  caption?: string;
};

type NewsArticle = {
  source: NewsSource;
  url: string;
  title: string;
  publishedAt: string;
  receivedAt: number;
  modifiedAt?: string;
  primaryCategory?: string;
  categories: string[];
  author?: string;
  keywords: string[];
  description?: string;
  summary?: string;
  media: NewsMedia[];
  content?: string;
  language?: string;
  copyright?: string;
};

type EnvelopeBase<
  TFamily extends 'tweet' | 'account' | 'charity' | 'control' | 'news',
  TOp extends string,
  TPayload extends object,
> = {
  v: 1;
  t: TFamily;
  op: TOp;
  id?: string;
  ts: number;
  d: TPayload;
};

type TweetContentEnvelope = EnvelopeBase<'tweet', 'content', TweetContent>;
type TweetMetaEnvelope = EnvelopeBase<'tweet', 'meta', TweetMeta>;
type TweetUpdateEnvelope = EnvelopeBase<'tweet', 'update', TweetUpdate>;
type TweetDeleteEnvelope = EnvelopeBase<'tweet', 'delete', TweetDeleteEvent>;
type TweetPinEnvelope = EnvelopeBase<'tweet', 'pin', TweetPinEvent>;
type TweetUnpinEnvelope = EnvelopeBase<'tweet', 'unpin', TweetUnpinEvent>;
type AccountProfileEnvelope = EnvelopeBase<
  'account',
  'profile_update',
  ProfileUpdateEvent
>;
type AccountFollowEnvelope = EnvelopeBase<'account', 'follow', FollowEvent>;
type AccountUnfollowEnvelope = EnvelopeBase<'account', 'unfollow', UnfollowEvent>;
type AccountAffiliateEnvelope = EnvelopeBase<
  'account',
  'affiliate_update',
  AffiliateUpdateEvent
>;
type AccountCalloutEnvelope = EnvelopeBase<'account', 'callout', CalloutEvent> & {
  id: string;
};
type AccountCalloutUpdateEnvelope = EnvelopeBase<
  'account',
  'callout_update',
  CalloutUpdateEvent
> & {
  id: string;
};
type AccountThesisEnvelope = EnvelopeBase<'account', 'thesis', FomoThesisEvent> & {
  id: string;
};
type AccountTokenVerifiedEnvelope = EnvelopeBase<
  'account',
  'token_verified',
  FomoTokenVerifiedEvent
> & {
  id: string;
};
type AccountEnvelope =
  | AccountProfileEnvelope
  | AccountFollowEnvelope
  | AccountUnfollowEnvelope
  | AccountAffiliateEnvelope
  | AccountCalloutEnvelope
  | AccountCalloutUpdateEnvelope
  | AccountThesisEnvelope
  | AccountTokenVerifiedEnvelope;
type CharityAddedEnvelope = EnvelopeBase<'charity', 'added', CharityAddedEvent> & {
  id: string;
};
type EmptyPayload = Record<string, never>;
type AuthPingCommand = { op: 'auth_ping' };
type AuthPongEnvelope = EnvelopeBase<'control', 'auth_pong', EmptyPayload>;
type TwitterHandlesResultEnvelope = EnvelopeBase<
  'control',
  'twitter_handles_result',
  TwitterHandlesResult
>;
type ControlEnvelope = AuthPongEnvelope | TwitterHandlesResultEnvelope;
type NewsEnvelope = EnvelopeBase<'news', 'article', NewsArticle> & {
  id: string;
};

type TweetStreamEnvelope =
  | TweetContentEnvelope
  | TweetMetaEnvelope
  | TweetUpdateEnvelope
  | TweetDeleteEnvelope
  | TweetPinEnvelope
  | TweetUnpinEnvelope
  | AccountEnvelope
  | CharityAddedEnvelope
  | NewsEnvelope
  | ControlEnvelope;

// Feed this router only values returned by a complete runtime decoder.
// See the reference consumer linked below for the executable decoder.
type DecodedEnvelope =
  | { kind: 'known'; event: TweetStreamEnvelope }
  | { kind: 'unknown' };

type ProtocolHandlers = {
  tweetContent: (event: TweetContentEnvelope) => void;
  tweetMeta: (event: TweetMetaEnvelope) => void;
  tweetUpdate: (event: TweetUpdateEnvelope) => void;
  tweetDelete: (event: TweetDeleteEnvelope) => void;
  tweetPin: (event: TweetPinEnvelope) => void;
  tweetUnpin: (event: TweetUnpinEnvelope) => void;
  accountProfileUpdate: (event: AccountProfileEnvelope) => void;
  accountFollow: (event: AccountFollowEnvelope) => void;
  accountUnfollow: (event: AccountUnfollowEnvelope) => void;
  accountAffiliateUpdate: (event: AccountAffiliateEnvelope) => void;
  accountCallout: (event: AccountCalloutEnvelope) => void;
  accountCalloutUpdate: (event: AccountCalloutUpdateEnvelope) => void;
  accountThesis: (event: AccountThesisEnvelope) => void;
  accountTokenVerified: (event: AccountTokenVerifiedEnvelope) => void;
  charityAdded: (event: CharityAddedEnvelope) => void;
  newsArticle: (event: NewsEnvelope) => void;
  authPong: (event: AuthPongEnvelope) => void;
  twitterHandlesResult: (event: TwitterHandlesResultEnvelope) => void;
  // Count and discard unknown input without logging its raw payload.
  unknown: () => void;
};

function assertNever(value: never): never {
  throw new Error('Unhandled validated envelope');
}

function route(decoded: DecodedEnvelope, handlers: ProtocolHandlers): void {
  if (decoded.kind === 'unknown') {
    handlers.unknown();
    return;
  }

  const event = decoded.event;
  switch (event.t) {
    case 'tweet':
      switch (event.op) {
        case 'content': handlers.tweetContent(event); return;
        case 'meta': handlers.tweetMeta(event); return;
        case 'update': handlers.tweetUpdate(event); return;
        case 'delete': handlers.tweetDelete(event); return;
        case 'pin': handlers.tweetPin(event); return;
        case 'unpin': handlers.tweetUnpin(event); return;
        default: return assertNever(event);
      }
    case 'account':
      switch (event.op) {
        case 'profile_update': handlers.accountProfileUpdate(event); return;
        case 'follow': handlers.accountFollow(event); return;
        case 'unfollow': handlers.accountUnfollow(event); return;
        case 'affiliate_update': handlers.accountAffiliateUpdate(event); return;
        case 'callout': handlers.accountCallout(event); return;
        case 'callout_update': handlers.accountCalloutUpdate(event); return;
        case 'thesis': handlers.accountThesis(event); return;
        case 'token_verified': handlers.accountTokenVerified(event); return;
        default: return assertNever(event);
      }
    case 'charity':
      handlers.charityAdded(event);
      return;
    case 'news':
      handlers.newsArticle(event);
      return;
    case 'control':
      switch (event.op) {
        case 'auth_pong': handlers.authPong(event); return;
        case 'twitter_handles_result': handlers.twitterHandlesResult(event); return;
        default: return assertNever(event);
      }
    default:
      return assertNever(event);
  }
}
```

## 准备生产环境

- 连接关闭或发生网络错误后，使用退避策略重连。
- 只连接文档列出的 TweetStream 端点，并使用上文说明的认证方式。路由由 TweetStream 处理，不需要基础设施专用 Header。
- 使用 `(platform, tweetId)` 作为帖子键，而不是消息信封 id。content 从 `author.platform` 读取 `platform`，后续操作从 `d.platform` 读取。
- 旧版操作缺少 `d.platform` 时，使用其中的 author platform；如果没有，则按 `twitter` 处理。
- 幂等应用每个 `tweet/update`。更新包含 `ref` 时，用该快照替换已存引用，并沿 `ref.subtweet` 读取下一条被引用的帖子。
- 只有对完整消息信封计算指纹后，才丢弃完全相同的重放。
- 把 `meta` 视为某条已路由推文的迟到富化信息。
- 将回执和投影存入带索引的持久化存储。根据回放窗口和存储预算制定保留或归档策略；不要删除待处理任务。
- 记录操作失败时不要写入 payload，并通过退避策略重试。使用相同的回执 ID 作为幂等键。
- 跟踪活跃 WebSocket 连接数，避免超过套餐限制。

## 测量信号延迟

在消息回调开始时记录本地接收时间。对于 X/Twitter 事件，解码 tweet snowflake 时间戳，测量从发布到接收的时间。主机时钟必须与 NTP 同步，wall-clock 结果才有意义。接收后的处理阶段使用单调时钟。

- 报告监控列表、客户端区域、UTC 测试时段、样本量、p50 和 p95。
- 记录重连、缺失事件、时钟同步状态和每条排除规则。
- 单独报告冷启动和重连后的样本，不要将它们混入热路径结果。
- 只有发布点、接收点、地理位置、样本和百分位边界一致时，才比较结果。

| 阶段 | 起点 | 终点 |
| --- | --- | --- |
| 发布到接收 | X snowflake 时间戳 | WebSocket 回调开始时记录的本地时间 |
| 接收到决策 | 本地 WebSocket 接收 | 策略与风控决策就绪 |
| 决策到交易场所确认 | 通过风控的决策 | 单独的交易场所响应或拒绝 |

### 测量 Snowflake 延迟

```typescript
type ContentEvent = {
  t?: string;
  op?: string;
  d?: {
    tweetId?: string;
  };
};

const TWITTER_EPOCH_MS = 1_288_834_974_657n;

function tweetIdToTimestampMs(tweetId: string) {
  const id = BigInt(tweetId);
  return Number((id >> 22n) + TWITTER_EPOCH_MS);
}

function measureSnowflakeLatency(tweetId: string, arrivedAtMs: number) {
  const tweetedAtMs = tweetIdToTimestampMs(tweetId);
  return arrivedAtMs - tweetedAtMs;
}

ws.on("message", (raw) => {
  const arrivedAtMs = Date.now();
  const event = JSON.parse(raw.toString()) as ContentEvent;
  const tweetId = event.d?.tweetId;

  if (event.t === "tweet" && event.op === "content" && tweetId) {
    const latencyMs = measureSnowflakeLatency(tweetId, arrivedAtMs);
    console.log("publication-to-receipt ms", latencyMs);
  }
});
```

## 构建参考消费者

构建自己的消费者时，这些页面分别讲解每个部分。先使用上方的分发器，再添加一种重连客户端、持久恢复和账号管理。

- [分发所有操作](https://tweetstream.io/zh-cn/docs/production-consumer#complete-router): 覆盖全部 18 种服务端操作的类型化处理程序。
- [使用 Node.js 重连](https://tweetstream.io/zh-cn/docs/connection-auth-heartbeat#connect): 区分连接升级失败，并通过退避策略重连。
- [使用 Python 重连](https://tweetstream.io/zh-cn/docs/connection-auth-heartbeat#python): 使用 Python 运行相同的连接循环。
- [使用 History 恢复](https://tweetstream.io/zh-cn/docs/envelope-delivery-recovery-errors#backfill): 用不透明 cursor 回放已存数据页。
- [查询 History](https://tweetstream.io/zh-cn/docs/history-api#request): 设置有界时间范围和固定筛选条件。
- [管理账号](https://tweetstream.io/zh-cn/docs/account-management#add-remove): 添加和移除受监控的 Twitter handle。