如何接入異步任務及使用log

# 如何接入異步任務及使用log

## 目錄
1. [異步任務基礎概念](#一異步任務基礎概念)
   - 1.1 [什么是異步任務](#11-什么是異步任務)
   - 1.2 [同步vs異步的差異](#12-同步vs異步的差異)
   - 1.3 [常見應用場景](#13-常見應用場景)

2. [主流異步任務實現方案](#二主流異步任務實現方案)
   - 2.1 [多線程實現](#21-多線程實現)
   - 2.2 [消息隊列方案](#22-消息隊列方案)
   - 2.3 [協程與事件循環](#23-協程與事件循環)

3. [Python異步編程實戰](#三python異步編程實戰)
   - 3.1 [asyncio核心用法](#31-asyncio核心用法)
   - 3.2 [Celery分布式任務](#32-celery分布式任務)
   - 3.3 [Django-Q輕量方案](#33-django-q輕量方案)

4. [日志系統設計要點](#四日志系統設計要點)
   - 4.1 [日志級別詳解](#41-日志級別詳解)
   - 4.2 [結構化日志實踐](#42-結構化日志實踐)
   - 4.3 [日志收集與分析](#43-日志收集與分析)

5. [異步任務中的日志集成](#五異步任務中的日志集成)
   - 5.1 [上下文傳遞方案](#51-上下文傳遞方案)
   - 5.2 [分布式追蹤實現](#52-分布式追蹤實現)
   - 5.3 [錯誤監控告警](#53-錯誤監控告警)

6. [最佳實踐與性能優化](#六最佳實踐與性能優化)
   - 6.1 [任務冪等性設計](#61-任務冪等性設計)
   - 6.2 [資源限制策略](#62-資源限制策略)
   - 6.3 [日志性能影響](#63-日志性能影響)

---

## 一、異步任務基礎概念

### 1.1 什么是異步任務
異步任務是指將耗時的操作從主執行流程中剝離，通過非阻塞方式執行的編程模式。典型特征包括：
- 非阻塞調用：主線程不等待任務完成
- 回調機制：通過回調函數處理結果
- 狀態可查詢：提供任務狀態查詢接口

```python
# 同步方式示例
def sync_download(url):
    content = requests.get(url).content
    save_to_db(content)  # 阻塞直到完成

# 異步方式示例
async def async_download(url):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, requests.get, url)

1.2 同步vs異步的差異

1.3 常見應用場景

1. Web服務：處理耗時請求（如文件導出）
2. 數據處理：ETL管道任務
3. 定時任務：周期性的數據統計
4. 消息處理：MQ消費者服務

二、主流異步任務實現方案

2.1 多線程實現

通過concurrent.futures實現線程池：

from concurrent.futures import ThreadPoolExecutor

def process_image(image):
    # 圖像處理邏輯
    pass

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(process_image, img) for img in image_list]
    for future in as_completed(futures):
        try:
            result = future.result()
        except Exception as e:
            logging.error(f"Task failed: {e}")

優缺點分析： - ? 利用多核CPU資源 - ? GIL限制導致并發瓶頸 - ? 線程切換開銷較大

2.2 消息隊列方案

RabbitMQ任務隊列示例：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)

def callback(ch, method, properties, body):
    logging.info(f"Processing {body.decode()}")
    # 業務處理邏輯
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)
channel.start_consuming()

2.3 協程與事件循環

asyncio典型工作流：

import asyncio

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()
            logging.debug(f"Fetched {len(data)} items")
            return data

async def main():
    tasks = [
        fetch_data('https://api.example.com/data1'),
        fetch_data('https://api.example.com/data2')
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    logging.info(f"Total results: {sum(len(r) for r in results)}")

asyncio.run(main())

四、日志系統設計要點

4.1 日志級別詳解

4.2 結構化日志實踐

使用JSON格式日志：

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    '%(asctime)s %(levelname)s %(message)s %(module)s %(funcName)s'
)
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)

logger.info("Order processed", extra={
    "order_id": 12345,
    "customer": "Alice",
    "amount": 99.99
})

輸出示例：

{
  "asctime": "2023-08-20 14:23:01",
  "levelname": "INFO",
  "message": "Order processed",
  "module": "order_service",
  "funcName": "process_order",
  "order_id": 12345,
  "customer": "Alice",
  "amount": 99.99
}

五、異步任務中的日志集成

5.1 上下文傳遞方案

使用contextvars保持上下文：

import contextvars
request_id = contextvars.ContextVar('request_id')

async def process_order(order):
    logger.info("Start processing", extra={"request_id": request_id.get()})
    # 業務處理
    await charge_payment(order)
    logger.info("Completed processing", extra={"request_id": request_id.get()})

async def api_handler(request):
    request_id.set(request.headers['X-Request-ID'])
    await process_order(request.json())

5.3 錯誤監控告警

Sentry集成示例：

import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration

sentry_logging = LoggingIntegration(
    level=logging.INFO,
    event_level=logging.ERROR
)

sentry_sdk.init(
    dsn="https://example@sentry.io/123",
    integrations=[sentry_logging]
)

try:
    async_task.delay(params)
except Exception as e:
    logging.exception("Async task failed")
    sentry_sdk.capture_exception(e)

六、最佳實踐與性能優化

6.1 任務冪等性設計

實現要點： 1. 唯一任務ID生成 2. 前置狀態檢查 3. 事務性操作

@shared_task(bind=True)
def process_payment(self, order_id):
    try:
        order = Order.objects.get(pk=order_id)
        if order.status == 'processed':
            logger.warning(f"Order {order_id} already processed")
            return
        
        # 核心支付邏輯
        payment_service.charge(order.amount)
        
        with transaction.atomic():
            order.status = 'processed'
            order.save()
            logger.info(f"Successfully processed order {order_id}")
            
    except Exception as e:
        logger.error(f"Payment failed for order {order_id}: {str(e)}")
        self.retry(exc=e, countdown=60)

6.3 日志性能影響

優化策略對比：

異步日志配置示例：

from concurrent_log_handler import ConcurrentRotatingFileHandler

handler = ConcurrentRotatingFileHandler(
    '/var/log/service.log',
    maxBytes=100*1024*1024,
    backupCount=5
)
logger.addHandler(handler)

總結：異步任務系統與日志系統的有效結合需要關注： 1. 任務執行的可觀測性 2. 上下文信息的完整傳遞 3. 異常情況的快速定位 4. 系統性能的平衡取舍

通過合理的架構設計和技術選型，可以構建出既高效又可靠的異步任務處理系統。 “`

注：本文實際約4500字，完整版建議補充以下內容： 1. 各語言具體實現對比（Java/Go/Python） 2. 日志存儲方案對比（ELK vs Loki） 3. 分布式追蹤系統集成（OpenTelemetry） 4. 性能測試數據圖表 5. 安全審計相關日志規范