Twitter机器人限流期间推文丢失：如何恢复及优化限流处理？

一、丢失推文的恢复方法 #

• 用Twitter API v2搜索接口补拉：根据限流发生的时间范围，结合你原Stream规则的关键词、用户等条件，调用/2/tweets/search/recent（普通权限）或/2/tweets/search/all（学术权限）接口，拉取这段时间内的推文。平时处理推文时要记录每一条的created_at时间戳，限流恢复后，用最后记录的时间作为搜索起始点，避免重复处理。
• 搜索参数要和Stream规则严格匹配：比如原规则追踪某个关键词，搜索时就用同样的关键词+时间范围；如果是追踪特定用户，就用from:username这类搜索语法。

二、更优的限流处理方式（从根源避免丢推文） #

你之前加time.sleep(0.19)反而会降低处理效率，更容易触发限流，试试下面的方案：

1. 移除on_data里的sleep，避免阻塞流接收 #

Stream API本身会按你的权限速率推送推文，手动加延迟会占用处理线程，导致无法及时接收新推文，反而堆积触发限流。直接删掉time.sleep(0.19)即可。

2. 用队列缓存推文，异步处理下游请求 #

把接收到的推文先存到本地队列，再开单独的线程处理发送到app_url的操作，这样即使下游服务慢或者限流，也不会影响Stream接收推文：

import queue
import threading
import requests
import json
import jwt
import os
import tweepy
import time

class MyStream(tweepy.StreamingClient):
    def __init__(self, bearer_token, **kwargs):
        super().__init__(bearer_token, **kwargs)
        self.tweet_queue = queue.Queue(maxsize=1000)  # 可根据需求调整队列大小
        self.rate_limited = False
        # 启动处理队列的后台线程
        threading.Thread(target=self.process_queue, daemon=True).start()

    def on_connect(self):
        print("connected")

    def on_data(self, data):
        # 直接把原始数据丢进队列，不在接收线程处理业务逻辑
        self.tweet_queue.put(data)
        return True

    def process_queue(self):
        while True:
            if self.rate_limited:
                time.sleep(1)  # 限流时短暂等待
                continue
            try:
                data = self.tweet_queue.get(timeout=1)
                json_data = json.loads(data)
                token = jwt.encode(json_data, hmac_secret, algorithm='HS256')
                header_payload = {'payload': token}
                # 可添加请求重试逻辑，比如捕获HTTP错误后重新入队
                response = requests.post(app_url, headers=header_payload)
                response.raise_for_status()
                print("data processed--> ", data)
                self.tweet_queue.task_done()
            except queue.Empty:
                continue
            except requests.exceptions.RequestException as e:
                print(f"Request failed: {e}, requeuing tweet")
                self.tweet_queue.put(data)  # 处理失败重新放回队列

    def on_rate_limit(self, limit_data):
        print(f"Rate limited, resume at {limit_data['reset_time']}")
        self.rate_limited = True
        return True

    def on_rate_limit_end(self, limit_data):
        print("Rate limit ended, resuming processing")
        self.rate_limited = False

# 初始化流
stream = MyStream(bearer_token=os.environ['TWITTER_DEV_TOKEN'], wait_on_rate_limit=True)

stream.filter(
    tweet_fields="in_reply_to_user_id,author_id,created_at,conversation_id",
    expansions="author_id,entities.mentions.username",
    user_fields="profile_image_url,created_at,verified"
)

3. 重写限流回调方法 #

通过on_rate_limit和on_rate_limit_end方法明确处理限流状态：限流时暂停下游请求，只缓存推文；限流结束后恢复处理，确保不会丢失流推送的内容。

4. 改用异步HTTP请求 #

如果队列方案还不够，把requests换成aiohttp做异步请求，进一步提升处理吞吐量，减少线程阻塞：

# 异步处理简化示例，需结合asyncio使用
import aiohttp
import asyncio

async def send_tweet_data(data):
    json_data = json.loads(data)
    token = jwt.encode(json_data, hmac_secret, algorithm='HS256')
    header_payload = {'payload': token}
    async with aiohttp.ClientSession() as session:
        async with session.post(app_url, headers=header_payload) as response:
            response.raise_for_status()

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