Python多进程场景下已检测Xbox用户名未正确追加到列表问题求助

你使用的multiprocessing多进程模式下，各个子进程会独立拷贝父进程的全部内存数据，你定义的全局Checked列表在每个子进程里都有独立的副本，不同进程之间的数据完全不互通。进程A把用户名加入自己的Checked列表，进程B根本感知不到，自然做不到全局去重。

方案1：改用多线程（最简便，适配IO密集型场景） #

你的脚本属于典型的IO密集型任务，大部分时间在等待请求返回，多线程完全可以满足并发需求，且线程之间共享同一块内存空间，全局Checked列表天然可以全局生效，不需要额外处理共享逻辑，只需要把multiprocessing.Process换成threading.Thread即可。

方案2：多进程下使用共享数据结构 #

如果一定要保留多进程架构，需要使用multiprocessing.Manager提供的共享列表/集合，同时加锁避免多进程并发写入的竞态问题：

• 用Manager().set()替代普通列表做去重，in查询复杂度从O(n)降到O(1)，性能提升明显
• 新增互斥锁，保证查询、写入共享集合的操作是原子性的，避免并发判断遗漏

• 原代码每次循环都会重复读取https.txt代理文件，完全没必要，程序启动时读取一次存入全局变量即可，减少IO开销
• 原异常捕获块未覆盖r.status_code的判断逻辑，请求失败后访问r变量会直接报错
• 原代码shtuf长度为36，用random.randrange(35)会漏掉最后一个元素z
• 原代码代理配置的键https://写法错误，requests的代理配置键不需要末尾斜杠，否则代理不会生效

import requests, random, string, threading
from threading import Lock

headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36'}
shtuf = (1, 2, 3, 4, 5, 6, 7, 8, 9, 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z')
Checked = set()
lock = Lock()
# 提前一次性读取代理列表
with open("https.txt", "r") as f:
    proxie_list = [proxy.strip() for proxy in f.readlines()]

def _3_letter_check_mixed():
    print('starting 3 letter check (mixed)')
    while True:
        user = random.choice(string.ascii_letters).lower() + str(shtuf[random.randrange(36)]).lower() + str(shtuf[random.randrange(36)]).lower()
        proxies = {
           'https': random.choice(proxie_list)
        }
        # 加锁保证查询和写入的原子性
        with lock:
            if user in Checked:
                continue
            Checked.add(user)
        try:
            r = requests.get(f'https://xboxgamertag.com/search/{user}', headers=headers, proxies=proxies, timeout=5)
            if r.status_code == 404:
                with open('usernames.txt', 'a') as t:
                    print(f'{user} : AVAILABLE')
                    t.write(user + '\n')
            else:
                print(f'{user} : taken')
        except Exception as e:
            print('REQUEST FAILED')

def _3_letter_check_letter_only():
    print('starting 3 letter check')
    while True:
        user = random.choice(string.ascii_letters).lower() + random.choice(string.ascii_letters).lower() + random.choice(string.ascii_letters).lower()
        proxies = {
           'https': random.choice(proxie_list)
        }
        with lock:
            if user in Checked:
                continue
            Checked.add(user)
        try:
            r = requests.get(f'https://xboxgamertag.com/search/{user}', headers=headers, proxies=proxies, timeout=5)
            if r.status_code == 404:
                with open('usernames.txt', 'a') as t:
                    print(f'{user} : AVAILABLE')
                    t.write(user + '\n')
            else:
                print(f'{user} : taken')
        except:
            print('REQUEST FAILED')
     
if __name__ == "__main__":
    threads = []
    for i in range(5):
        t = threading.Thread(target=_3_letter_check_mixed)
        t.start()
        threads.append(t)
    for i in range(5):
        t = threading.Thread(target=_3_letter_check_letter_only)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

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