Celery Playwright异步爬虫

安装环境

安装Python包

# 包
pip3 install celery
pip3 redis
pip3 playwright
# 下载chromium浏览器（可选）
playwright install

安装redis

参考菜鸟教程

代码

在程序根目录建立test.py文件

import asyncio
import time
from celery import Celery
from celery.result import AsyncResult
from playwright.async_api import async_playwright
app = Celery("mytasks", backend="redis://localhost", broker="redis://localhost:6379/0")
# async语法的异步爬虫耗时任务
async def async_func():
    playwright = await async_playwright().start()
    browser = await playwright.chromium.launch(headless=False)
    page = await browser.new_page()
    await page.goto("http://zhihu.com")
    title = await page.title()
    await asyncio.sleep(10)
    await browser.close()
    return title
@app.task(name="mytasks")
def io_func():
    # 核心代码asyncio.run
    return asyncio.run(async_func())
def main():
    result = io_func.delay()
    return result.id
# 循环访问获取异步任务结果
def get_result(task_id):
    res = AsyncResult(task_id)
    while True:
        print(res.status)
        if res.ready():
            break
        time.sleep(1)
    return res.result
if __name__ == "__main__":
    task_id = main()
    print(task_id)
    res = get_result(task_id)
    print(res)

启动celery

打开终端，进入程序根目录，执行以下代码启动celery

celery -A test worker -B --loglevel=info

开始测试

打开终端，进入程序根目录，执行

python3 test.py