async def main():    # 1. 初始化Redis连接池    await RedisManager.initialize()    RedisManager.start_daemon()    # 2. 初始化MySQL连接池    mysql_manager = MySQLManager(MYSQL_CONF)    await mysql_manager.init_pool()    # 3. 启动资源监控（后台任务）    monitor_task = asyncio.create_task(        monitor.monitor_process_resources(interval=5)    )    # 4. 创建消费者协程    consumers = []    for t in TASKS:        workers = t.get("workers", 1)        for i in range(workers):            consumers.append(consume_task(...))    # 5. 并发执行所有消费者    await asyncio.gather(*consumers)

TASKS = [    {        "key": "amacfund:fund_detail",        "func": amac_fund_crawler.get_fund_detail,        "log": "基金详情",        "field": "fund_id",        "workers": 20,  # 20个并发协程        "limit": 20,    # 信号量限制        "need_paging": False    }]

async def main():    scheduler = AsyncIOScheduler()    # 添加定时任务（每天7:00执行）    scheduler.add_job(        run_task,        CronTrigger(hour=7, minute=0),        id='amacfund_crawler',        name='基金数据采集'    )    scheduler.start()    await asyncio.Event().wait()

# 每天7:00CronTrigger(hour=7, minute=0)# 每小时CronTrigger(minute=0)# 每5分钟CronTrigger(minute='*/5')# 工作日9:00CronTrigger(day_of_week='mon-fri', hour=9, minute=0)

async with sem:  # 信号量控制并发    result = await asyncio.wait_for(        crawler_func(**params),        timeout=timeout    )

if task_key not in task_locks:    task_locks[task_key] = asyncio.Semaphore(limit)

机器1 (runner_batch.py)  ──┐                           ├──→ Redis队列 ←──→ MySQL机器2 (runner_batch.py)  ──┤                           │机器3 (runner_batch.py)  ──┘

# 机器1python runner_batch.py# 机器2python runner_batch.py# 机器3python runner_batch.py```

async def shutdown():    stop_flag = True    # 未完成任务放回队列    for task_key, record in running_tasks:        await RedisManager.lpush(retry_key, record)    # 关闭数据库连接    mysql_manager.mysql.close()    await mysql_manager.mysql.wait_closed()

# 重试3次，指数退避for attempt inrange(max_retries):    try:        result = await crawler_func(**params)        break    except Exception as e:        if attempt < max_retries:            await asyncio.sleep(retry_delay * attempt)# 失败后放入重试队列if attempt >= max_retries:    await RedisManager.lpush(retry_key, json.dumps(params))

# 2小时无新任务，自动停止if time_since_last_activity > QUEUE_TIMEOUT:    task_stop_flags[task_key] = True    break

async def monitor_process_resources(interval=5):    process = psutil.Process(os.getpid())    while True:        # 内存使用        memory_mb = process.memory_info().rss / 1024 / 1024        # CPU使用率        cpu_usage = (current_cpu_time - last_cpu_time) /                    (current_time - last_time) * 100        # 线程数        num_threads = process.num_threads()        logging.info(f"{timestamp} | 内存:{memory_mb:.2f}MB | CPU:{cpu_usage:.1f}% | 线程:{num_threads}")        await asyncio.sleep(interval)

@save_into_mysql(table_name="fund_info", unique_key="pmid")@parse_response(parse_fund_detail, extra_args=["self", "**kwargs"])@async_request(method="GET", delay=0.1, if_proxy=True)async def get_fund_detail(self, fund_id, **kwargs):    url = AMAC_FUND_DETAIL_API.format(fund_id=fund_id)    return {        "url": url,        "headers": {},        "crawler": self    }

1. get_fund_detail() 返回请求参数   ↓2. @async_request 发送HTTP请求   ↓3. @parse_response 解析响应数据   ↓4. @save_into_mysql 保存到数据库

async def _process_single_row(conn, cursor, row):    # 1. 检查记录是否存在    check_sql = f"SELECT 1 FROM {table} WHERE {unique_key} = %s"    exists = await cursor.fetchone()    if exists:        # 2. 存在则UPDATE        update_sql = f"UPDATE {table} SET ... WHERE {unique_key}=%s"        await cursor.execute(update_sql, values)    else:        # 3. 不存在则INSERT        insert_sql = f"INSERT INTO {table} (...) VALUES (...)"        await cursor.execute(insert_sql, values)

# 先尝试批量插入success = await _execute_batch_all()if not success and enable_fallback:    # 降级：逐条插入    for row in rows:        if not await _execute_single(row):            failed_records.append(row)

async def _save_failed_records(table, failed_records):    filename = f"logs/failed_records/{table}_{timestamp}.json"    with open(filename, 'w', encoding='utf-8') as f:        json.dump({            "table": table,            "count": len(failed_records),            "records": failed_records        }, f, ensure_ascii=False, indent=2)

async def parse_fund_list(data, crawler, **kwargs):    records = data.get('content', [])    result = []    for record in records:        fund_info = {            'fund_id': record.get("id"),            'fund_name': record.get("fundName"),            ...        }        result.append(fund_info)        # 推送详情任务到队列        await crawler.redis.lpush(            "amacfund:fund_detail",            json.dumps(fund_info)        )    return result

async def parse_fund_detail(response, crawler, **kwargs):    soup = BeautifulSoup(response, 'html.parser')    fund_info = {}    rows = soup.select('div.info-body table.table tr')    for row in rows:        cells = row.find_all('td')        title = cells[0].get_text(strip=True)        value = cells[1].get_text(strip=True)        if title == '基金类型':            fund_info['fund_type'] = value        ...    return fund_info

logging.info(f"✅ 采集成功 | 基金ID: {fund_id} | 耗时: {elapsed}s")logging.error(f"❌ 采集失败 | 基金ID: {fund_id} | 错误: {error}")logging.warning(f"⏰ 队列超时 | 队列: {task_key} | 超时: {timeout}s")

# INFO级别：正常日志logging.basicConfig(level=logging.INFO)# ERROR级别：错误日志（单独文件）setup_error_logger()

{  "table": "fund_info",  "timestamp": "20260212_143025",  "count": 5,  "records": [    {"fund_id": "123", "fund_name": "xxx"}  ]}

redis-cli LPUSH amacfund:amac_fund_list '{"page": 0, "page_size": 20}'

python runner_batch.py

消费列表任务  ↓解析列表页（20条基金）  ↓推送20个详情任务到队列  ↓20个协程并发消费详情任务  ↓解析详情页  ↓保存到MySQL  ↓继续消费下一页