如何使此代码运行得更好/更快（线程或多处理）？怎么做？

Question

我有一个简单的机器人，它使用 cookie 登录网站并检查商品的价格，如果该价格符合我设置的价格，它就会购买该商品。

我正在寻找提高此机器人速度的方法。我真的不知道在这种情况下多处理是否会让这个机器人更快。

我也在寻找提高效率的方法。

session = requests.session()
session.cookies["cookie"] = ""

log_in = session.get("https://www.example.com")
if log_in.status_code == 200:
    print("Logged In")
else:
    raise ValueError("Invalid Cookie")

crsf_token = ""

def token():
    global crsf_token
    while True:
        crsf_token = re.search(r"<script>XsrfToken.setToken\('(.*?)'\);</script>", session.get('https://www.example.com').text).group(1)
        time.sleep(5)

def _cthread():
    while True:
        try:
            req = session.get(f"https://www.example.com/productID")
            if req.status_code == 429:
                time.sleep(5)
                continue

            for i in req.json()["data"]["Sellers"]:
                if i["Price"] <= 300:
                    session.post(f'https://www.example.com&expectedPrice={i["Price"]}&expectedSellerID={i["SellerId"]}&userAssetID={i["UserAssetId"]}', headers={"X-CSRF-TOKEN": crsf_token})
        except requests.urllib3.exceptions.ConnectTimeoutError as E:
            pass


while True:
    threading.Thread(target=_cthread).start()
    threading.Thread(target=token).start()

我在这方面没有取得太大的成功，但它现在确实有效。

Answer 1

假设您不受带宽限制，并且该站点不会一次关闭太多提交的 POSTs，您可能会通过线程化您的 post 调用获得一点好处，更改：

        for i in req.json()["data"]["Sellers"]:
            if i["Price"] <= 300:
                session.post(f'https://www.example.com&expectedPrice={i["Price"]}&expectedSellerID={i["SellerId"]}&userAssetID={i["UserAssetId"]}', headers={"X-CSRF-TOKEN": crsf_token})

到：

        allposts = [f'https://www.example.com&expectedPrice={i["Price"]}&expectedSellerID={i["SellerId"]}&userAssetID={i["UserAssetId"]}'
                    for i in req.json()["data"]["Sellers"] if i["Price"] <= 300]
        if allposts:
            with Pool() as pool:
                pool.map(partial(session.post, headers={"X-CSRF-TOKEN": crsf_token}), allposts)

将以下导入添加到文件顶部：

from multiprocessing.dummy import Pool  # Gets thread based worker Pool class
from functools import partial           # Lets you bind headers up front so you only need to pass URL to post

为避免过多的线程产生，您可以在该函数的循环外创建pool，而不是仅在allposts 非空时按需创建。

我还建议从代码的顶层删除while True:； token 和 _cthread 已经是无限循环，因此拥有这两个循环意味着产生无限数量的线程，每个线程都永远运行，而您实际上只需要两个持久线程。

Answer 2

尝试从函数token()和_cthread()中删除while True