优化Instagram API的getfollowers函数：用多进程缩短耗时

首先帮你纠正原代码里的一个小问题：你自己定义了getTotalFollowers函数，但主函数里却调用了api.getTotalFollowers(user_id)——原库的InstagramAPI类并没有这个内置方法，会导致报错，应该改成followers = getTotalFollowers(user_id)。

接下来聊聊优化思路：你的原代码是同步串行请求，每次必须等上一页的粉丝数据返回后，才能请求下一页，大部分时间都浪费在等待API响应上（属于IO密集型任务）。要把耗时从3秒压缩到1秒左右，我们可以通过优化请求的阻塞逻辑来实现，下面给你两个新手友好的可行方案：

方案一：用concurrent.futures.ThreadPoolExecutor优化（上手简单） #

虽然分页请求必须依赖上一页的next_max_id、无法完全并行，但我们可以用线程池来减少IO阻塞的开销——线程池会在等待API响应时释放CPU资源，避免程序一直卡在等待状态。

修改后的代码如下：

from InstagramAPI import InstagramAPI
import time
from concurrent.futures import ThreadPoolExecutor

def fetch_single_page(api, user_id, maxid=''):
    """单独请求一页粉丝数据"""
    api.getUserFollowers(user_id, maxid=maxid)
    return api.LastJson.get('users', []), api.LastJson.get('next_max_id', '')

def getTotalFollowers(user_id, api):
    followers = []
    next_max_id = ''
    # 线程数设为2-3即可，避免触发Instagram限流
    with ThreadPoolExecutor(max_workers=2) as executor:
        while next_max_id is not None:
            # 提交单页请求任务到线程池
            future = executor.submit(fetch_single_page, api, user_id, next_max_id)
            page_followers, next_max_id = future.result()
            followers.extend(page_followers)
            # 加极短延迟，避免请求过于频繁
            time.sleep(0.1)
    return followers

if __name__ == "__main__":
    # 替换成你的Instagram账号密码
    api = InstagramAPI("your_username","your_password")
    api.login()
    user_id = 6839180758
    start_time = time.time()
    followers = getTotalFollowers(user_id, api)
    print('Number of followers:', len(followers))
    # 简化打印避免输出过长
    print('Follower details count:', len(followers))
    print(f"\n\nTotal time required: {time.time() - start_time:.2f} seconds")

方案说明： #

1. 把单页请求封装成独立函数，用线程池执行，减少同步阻塞的等待时间
2. 线程数不要设太高，避免触发Instagram的请求频率限制（会被临时限流甚至封号）
3. 加入极短延迟，进一步降低被限流的风险

方案二：用asyncio+aiohttp重构（效率更高） #

如果你想进一步压缩耗时，可以用异步HTTP库aiohttp重构请求逻辑——原InstagramAPI库基于同步requests，阻塞时间较长，异步请求能把IO等待的开销降到最低，大概率能把耗时压到1秒左右。这个方案需要一点异步基础，但代码逻辑清晰，新手也能快速上手：

import asyncio
import aiohttp
import time

class AsyncInstagramAPI:
    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.session = aiohttp.ClientSession()
        self.csrf_token = ''

    async def login(self):
        """异步登录获取会话权限"""
        # 先获取首页的CSRF Token
        async with self.session.get("https://www.instagram.com/") as resp:
            html = await resp.text()
            self.csrf_token = html.split('csrf_token":"')[1].split('"')[0]
        
        # 发送登录请求
        login_url = "https://www.instagram.com/accounts/login/ajax/"
        payload = {'username': self.username, 'password': self.password}
        headers = {
            'X-CSRFToken': self.csrf_token,
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
        }
        async with self.session.post(login_url, data=payload, headers=headers) as resp:
            result = await resp.json()
            if not result.get('authenticated'):
                raise Exception("登录失败，请检查账号密码")
            print("登录成功")

    async def fetch_single_page(self, target_user_id, maxid=''):
        """异步请求一页粉丝数据"""
        url = f"https://www.instagram.com/api/v1/friendships/{target_user_id}/followers/"
        params = {'max_id': maxid} if maxid else {}
        headers = {
            'X-CSRFToken': self.csrf_token,
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
        }
        async with self.session.get(url, params=params, headers=headers) as resp:
            data = await resp.json()
            return data.get('users', []), data.get('next_max_id', None)

    async def get_total_followers(self, target_user_id):
        followers = []
        next_max_id = ''
        while next_max_id is not None:
            page_followers, next_max_id = await self.fetch_single_page(target_user_id, next_max_id)
            followers.extend(page_followers)
            await asyncio.sleep(0.05)  # 限流延迟
        return followers

async def main():
    api = AsyncInstagramAPI("your_username", "your_password")
    await api.login()
    target_user_id = 6839180758
    start_time = time.time()
    followers = await api.get_total_followers(target_user_id)
    print('Number of followers:', len(followers))
    print(f"\n\nTotal time required: {time.time() - start_time:.2f} seconds")
    await api.session.close()

if __name__ == "__main__":
    asyncio.run(main())

方案说明： #

1. 自己封装了异步版的Instagram API，用aiohttp发送非阻塞请求，大幅降低IO等待时间
2. 同样加入限流延迟，避免触发Instagram的反爬机制
3. 这个方案的效率提升最明显，适合想深入了解异步编程的新手尝试

重要提醒： #

Instagram对第三方API请求的频率限制非常严格，不要过度并行请求，否则会被临时限制访问甚至封号。如果有条件，尽量使用官方开放的API（需要申请权限）。

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