跳到文档正文

构建生产环境消费者

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

运行完整示例
在聊天中打开
按产品代码校对:2026年9月29日

通过退避策略重连

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

分发所有服务端操作

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

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

完整 TypeScript 分发器

显示完整定义(632 行)
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);
  }
});

构建参考消费者

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