node.js使用twitter-api-v2如何同时运行多用户独立推文流

开发适配Discord的Twitter集成功能过程中遇到以下问题：

• 最初调用search recent tweets接口时触发速率限制
• 尝试切换为流式接口方案后，无法实现通过命令新增多个用户、同时为这些用户开启推文流的需求
• 现有代码在单用户场景下运行正常，但Twitter流式接口的规则会持久化保存，暂未找到运行多个独立流实例的可行方法

现有实现代码如下：

const { ETwitterStreamEvent, TweetStream, TwitterApi, ETwitterApiError } = require('twitter-api-v2');

const client = new TwitterApi(process.env["BEARER_TOKEN"]);

async function startStream(userName){
  const stream = await client.v2.searchStream();
  await client.v2.updateStreamRules({
  add: [
    { value: `from:${userName}`, tag: userName },
  ],
});
  
  stream.on(
    ETwitterStreamEvent.ConnectionError,
    err => console.log('Connection error!', err),
  );
  
  stream.on(
    ETwitterStreamEvent.ConnectionClosed,
    () => console.log('Connection has been closed.'),
  );
  
  stream.on(
    ETwitterStreamEvent.Data,
    eventData => console.log(eventData),
  );
  
  stream.autoReconnect = true;
}

核心诉求：找到可以同时为不同用户运行多个独立推文流的可行方案。

首先明确Twitter API v2过滤流的核心限制：单个Bearer Token仅允许同时维持1个活跃的搜索流连接，你不需要、也不能在单Token下开多个独立流实例，当前代码每次调用startStream就新建一个流连接的写法本身就不符合API设计规范，会直接触发连接数限制报错。

正确实现多用户订阅推文流的逻辑如下：

• 全局只初始化1个持久化的流连接，服务启动时就建立连接，不要每次新增订阅用户都新建流
• 本地维护一份订阅关系映射表，记录每个Discord频道/用户订阅的Twitter账号
• 新增/取消订阅Twitter用户时，不要操作流连接，只需要调用updateStreamRules接口，新增或删除对应的from:用户名规则即可；每次操作规则前先拉取当前已存在的持久化规则列表，避免重复添加相同规则触发报错
• 流连接收到推文数据时，根据返回数据里携带的匹配规则tag（就是你添加规则时传入的和用户名一致的tag字段），从本地订阅映射表里找到对应的Discord推送目标，把推文内容转发过去即可

修正后的核心实现参考：

const { ETwitterStreamEvent, TwitterApi } = require('twitter-api-v2');
const client = new TwitterApi(process.env["BEARER_TOKEN"]);
// 全局唯一流实例
let activeStream = null;
// 本地订阅映射：key为Twitter用户名，value为需要推送的Discord频道/用户ID列表
const subscribeMap = new Map();

// 服务启动时初始化唯一流
async function initSingleStream() {
  // 先拉取已存在的持久化规则，避免重复添加
  const existingRules = await client.v2.streamRules();
  const existingRuleValues = new Set(existingRules.data?.map(r => r.value) || []);
  // 把本地已有的订阅规则补上
  const rulesToAdd = [];
  for (const userName of subscribeMap.keys()) {
    const ruleValue = `from:${userName}`;
    if (!existingRuleValues.has(ruleValue)) {
      rulesToAdd.push({ value: ruleValue, tag: userName });
    }
  }
  if (rulesToAdd.length) {
    await client.v2.updateStreamRules({ add: rulesToAdd });
  }
  // 建立唯一流连接
  activeStream = await client.v2.searchStream({
    expansions: ['author_id'],
    'tweet.fields': ['created_at', 'text']
  });
  activeStream.autoReconnect = true;

  activeStream.on(ETwitterStreamEvent.ConnectionError, err => console.log('Stream connection error:', err));
  activeStream.on(ETwitterStreamEvent.ConnectionClosed, () => console.log('Stream connection closed, will auto-reconnect'));
  // 处理收到的推文，按tag分发到对应Discord目标
  activeStream.on(ETwitterStreamEvent.Data, async tweetData => {
    const matchedTag = tweetData.matching_rules?.[0]?.tag;
    if (!matchedTag || !subscribeMap.has(matchedTag)) return;
    const pushTargets = subscribeMap.get(matchedTag);
    // 这里写Discord推送逻辑，遍历pushTargets发消息即可
    console.log(`Push tweet from ${matchedTag} to targets:`, pushTargets, tweetData);
  });
}

// 新增订阅的方法，不需要新建流
async function addSubscription(userName, discordTargetId) {
  if (!subscribeMap.has(userName)) {
    subscribeMap.set(userName, []);
    // 给唯一流加新规则
    await client.v2.updateStreamRules({
      add: [{ value: `from:${userName}`, tag: userName }]
    });
  }
  const targets = subscribeMap.get(userName);
  if (!targets.includes(discordTargetId)) {
    targets.push(discordTargetId);
  }
}

// 取消订阅的方法
async function removeSubscription(userName, discordTargetId) {
  if (!subscribeMap.has(userName)) return;
  const targets = subscribeMap.get(userName);
  const targetIndex = targets.indexOf(discordTargetId);
  if (targetIndex > -1) targets.splice(targetIndex, 1);
  // 如果该Twitter用户已经没有任何订阅者，删除对应流规则
  if (targets.length === 0) {
    subscribeMap.delete(userName);
    const existingRules = await client.v2.streamRules();
    const targetRule = existingRules.data?.find(r => r.tag === userName);
    if (targetRule) {
      await client.v2.updateStreamRules({
        delete: { ids: [targetRule.id] }
      });
    }
  }
}

// 服务启动时初始化流
initSingleStream().catch(console.error);

注意：Twitter API v2基础版权限单流最多支持添加25条规则，单条规则可通过from:userA OR from:userB OR from:userC的格式最多匹配1000个用户，足够覆盖绝大多数Discord机器人的使用场景，完全不会触发之前搜索接口的速率限制。如果你的订阅用户量确实超过单Token的规则上限，才需要申请多个独立的Twitter开发者凭证，用不同的Bearer Token启动多个流实例做订阅分片，普通场景不需要这么做。

内容的提问来源于stack exchange，提问作者Nik Malmberg