# 基本的多進(jìn)程創(chuàng)建示例
from multiprocessing import Process

def worker(task_id):
    print(f"處理任務(wù) {task_id}")

if __name__ == '__main__':
    processes = []
    for i in range(3):
        p = Process(target=worker, args=(i,))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()

class AsyncResultStorage:
    """異步結(jié)果存儲器"""
    
    def __init__(self, storage_file="async_results.json", manager=None):
        self.storage_file = storage_file
        # 使用傳入的manager創(chuàng)建共享列表和鎖
        if manager:
            self.results = manager.list()
            self.lock = manager.Lock()
        else:
            self.results = []
            self.lock = multiprocessing.Lock()

with ProcessPoolExecutor(max_workers=4) as executor:
    # 提交所有任務(wù)
    future_to_task = {
        executor.submit(send_request, task_id, status_queue, storage): task_id
        for task_id in range(num_tasks)
    }
    
    # 使用as_completed獲取完成的任務(wù)
    for future in as_completed(future_to_task):
        task_id = future_to_task[future]
        try:
            result = future.result(timeout=30)
        except Exception as e:
            logger.error(f"任務(wù) {task_id} 執(zhí)行異常: {e}")

def monitor_status(status_queue, total_tasks, storage):
    """增強(qiáng)的監(jiān)控函數(shù)，支持實(shí)時(shí)統(tǒng)計(jì)和結(jié)果查詢"""
    completed = failed = 0
    start_time = time.time()
    
    while completed + failed < total_tasks:
        if not status_queue.empty():
            status_info = status_queue.get(timeout=1)
            # 更新進(jìn)度狀態(tài)...
            
        # 進(jìn)度條顯示邏輯
        progress = (completed + failed) / total_tasks
        bar_length = 30
        filled_length = int(bar_length * progress)
        bar = '█' * filled_length + '?' * (bar_length - filled_length)

# 創(chuàng)建管理器共享隊(duì)列
manager = multiprocessing.Manager()
status_queue = manager.Queue()

# 工作進(jìn)程發(fā)送狀態(tài)
status_queue.put({
    'task_id': task_id,
    'status': scenario['status'],
    'message': scenario['message'],
    'timestamp': datetime.now().isoformat()
})

# 使用Manager創(chuàng)建共享存儲
manager = multiprocessing.Manager()
storage = AsyncResultStorage(manager=manager)

# 多個(gè)進(jìn)程可以安全訪問共享數(shù)據(jù)
with storage.lock:
    storage.results.append(new_result)

def send_request(task_id, status_queue, storage):
    try:
        # 任務(wù)邏輯...
    except Exception as e:
        error_msg = f"任務(wù) {task_id} 執(zhí)行異常: {str(e)}"
        logger.error(error_msg)
        
        # 存儲異常結(jié)果，確保不會丟失失敗記錄
        storage.add_result(
            task_id=task_id,
            status='Failed',
            message=error_msg,
            response_data=None
        )

# 任務(wù)執(zhí)行超時(shí)控制
for future in as_completed(future_to_task):
    try:
        result = future.result(timeout=30)  # 30秒超時(shí)
    except TimeoutError:
        logger.error("任務(wù)執(zhí)行超時(shí)")
        # 標(biāo)記任務(wù)為失敗，釋放資源

# 根據(jù)任務(wù)類型動(dòng)態(tài)調(diào)整進(jìn)程數(shù)
if task_type == "cpu_intensive":
    workers = multiprocessing.cpu_count()
elif task_type == "io_intensive":
    workers = multiprocessing.cpu_count() * 2
else:
    workers = 4

with ProcessPoolExecutor(max_workers=4) as executor:
    # 執(zhí)行任務(wù)...
# 池自動(dòng)清理，無需手動(dòng)調(diào)用shutdown

# 與Celery集成的示例
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def distributed_task(task_id):
    # 在分布式環(huán)境中執(zhí)行任務(wù)
    return send_request(task_id, status_queue, storage)

import multiprocessing
import requests
import time
import random
import json
import logging
from datetime import datetime, timedelta
from concurrent.futures import ProcessPoolExecutor, as_completed

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class AsyncResultStorage:
    """異步結(jié)果存儲器"""

    def __init__(self, storage_file="async_results.json", manager=None):
        self.storage_file = storage_file
        # 使用傳入的manager創(chuàng)建共享列表和鎖，如果沒有提供manager則創(chuàng)建普通對象
        if manager:
            self.results = manager.list()
            self.lock = manager.Lock()
        else:
            self.results = []
            self.lock = multiprocessing.Lock()

    def add_result(self, task_id, status, message, response_data=None, timestamp=None):
        """添加結(jié)果到存儲（線程安全）"""
        if timestamp is None:
            timestamp = datetime.now().isoformat()

        result = {
            'task_id': task_id,
            'status': status,
            'message': message,
            'response_data': response_data,
            'timestamp': timestamp,
            'process_id': multiprocessing.current_process().pid
        }

        # 使用鎖確保線程安全
        with self.lock:
            self.results.append(result)

        # 異步寫入文件（實(shí)際應(yīng)用中可替換為數(shù)據(jù)庫存儲）
        self._async_save_to_file(result)

        return result

    def _async_save_to_file(self, result):
        """異步保存結(jié)果到文件"""
        try:
            with open(self.storage_file, 'a', encoding='utf-8') as f:
                f.write(json.dumps(result, ensure_ascii=False) + '\n')
        except Exception as e:
            logger.error(f"保存結(jié)果到文件失敗: {e}")

    def get_results_by_status(self, status):
        """根據(jù)狀態(tài)篩選結(jié)果"""
        with self.lock:
            return [r for r in self.results if r['status'] == status]

    def get_statistics(self):
        """獲取統(tǒng)計(jì)信息"""
        with self.lock:
            total = len(self.results)
            completed = len([r for r in self.results if r['status'] == "Completed"])
            failed = len([r for r in self.results if r['status'] == "Failed"])

        return {
            'total_tasks': total,
            'completed': completed,
            'failed': failed,
            'success_rate': completed / total if total > 0 else 0
        }


def send_request(task_id, status_queue, storage):
    """改進(jìn)的請求發(fā)送函數(shù)，支持結(jié)果存儲"""
    logger.info(f"任務(wù) {task_id} 開始執(zhí)行，進(jìn)程ID: {multiprocessing.current_process().pid}")

    try:
        # 模擬不同的請求類型和參數(shù)
        request_types = ['GET', 'POST']
        request_type = random.choice(request_types)

        # 模擬請求延遲
        delay = random.uniform(0.5, 3.0)
        time.sleep(delay)

        # 模擬不同的響應(yīng)情況
        response_scenarios = [
            {'status': 'Completed', 'message': '請求成功', 'data': {'score': random.randint(1, 100)}},
            {'status': 'Completed', 'message': '請求成功', 'data': {'result': 'ok'}},
            {'status': 'Failed', 'message': '請求超時(shí)', 'data': None},
            {'status': 'Failed', 'message': '服務(wù)器錯(cuò)誤', 'data': None}
        ]

        scenario = random.choice(response_scenarios)

        # 存儲詳細(xì)結(jié)果
        storage.add_result(
            task_id=task_id,
            status=scenario['status'],
            message=scenario['message'],
            response_data=scenario['data']
        )

        # 通知狀態(tài)隊(duì)列（用于實(shí)時(shí)監(jiān)控）
        status_queue.put({
            'task_id': task_id,
            'status': scenario['status'],
            'message': scenario['message'],
            'timestamp': datetime.now().isoformat()
        })

        logger.info(f"任務(wù) {task_id} 完成，狀態(tài): {scenario['status']}")

        return scenario['status']

    except Exception as e:
        error_msg = f"任務(wù) {task_id} 執(zhí)行異常: {str(e)}"
        logger.error(error_msg)

        # 存儲異常結(jié)果
        storage.add_result(
            task_id=task_id,
            status='Failed',
            message=error_msg,
            response_data=None
        )

        status_queue.put({
            'task_id': task_id,
            'status': 'Failed',
            'message': error_msg,
            'timestamp': datetime.now().isoformat()
        })

        return 'Failed'


def format_time_delta(seconds):
    """格式化時(shí)間差"""
    if seconds < 60:
        return f"{seconds:.1f}s"
    elif seconds < 3600:
        minutes = seconds / 60
        return f"{minutes:.1f}m"
    else:
        hours = seconds / 3600
        return f"{hours:.1f}h"


def monitor_status(status_queue, total_tasks, storage):
    """增強(qiáng)的監(jiān)控函數(shù)，支持實(shí)時(shí)統(tǒng)計(jì)和結(jié)果查詢"""
    completed = failed = 0
    last_report_time = time.time()
    start_time = time.time()
    
    # 進(jìn)度條相關(guān)變量
    last_progress_update = time.time()

    logger.info(f"監(jiān)控進(jìn)程啟動(dòng)，共監(jiān)控 {total_tasks} 個(gè)任務(wù)")

    while completed + failed < total_tasks:
        try:
            # 非阻塞獲取狀態(tài)信息
            if not status_queue.empty():
                status_info = status_queue.get(timeout=1)
                task_id = status_info['task_id']

                print(
                    f"[{status_info['timestamp']}] 任務(wù) {task_id} 狀態(tài): {status_info['status']} - {status_info['message']}")

                if status_info['status'] == "Completed":
                    completed += 1
                elif status_info['status'] == "Failed":
                    failed += 1
                    
                # 更新進(jìn)度條顯示
                current_time = time.time()
                if current_time - last_progress_update >= 1:  # 每秒更新一次進(jìn)度條
                    processed_tasks = completed + failed
                    progress = processed_tasks / total_tasks
                    percentage = progress * 100
                    
                    # 計(jì)算 ETA
                    elapsed_time = current_time - start_time
                    if processed_tasks > 0:
                        avg_time_per_task = elapsed_time / processed_tasks
                        remaining_tasks = total_tasks - processed_tasks
                        eta_seconds = avg_time_per_task * remaining_tasks
                        eta_formatted = format_time_delta(eta_seconds)
                        avg_time_formatted = format_time_delta(avg_time_per_task)
                    else:
                        eta_formatted = "未知"
                        avg_time_formatted = "未知"
                    
                    # 顯示進(jìn)度條
                    bar_length = 30
                    filled_length = int(bar_length * progress)
                    bar = '█' * filled_length + '?' * (bar_length - filled_length)
                    
                    print(f"\r進(jìn)度: |{bar}| {percentage:.1f}% 完成 ({processed_tasks}/{total_tasks}), "
                          f"平均耗時(shí): {avg_time_formatted}, ETA: {eta_formatted}", end='', flush=True)
                    last_progress_update = current_time

            else:
                time.sleep(0.1)  # 短暫休眠避免忙等待

            # 每5秒報(bào)告一次統(tǒng)計(jì)信息
            current_time = time.time()
            if current_time - last_report_time >= 5:
                stats = storage.get_statistics()
                logger.info(
                    f"任務(wù)進(jìn)度: {completed + failed}/{total_tasks} "
                    f"(成功: {stats['completed']}, 失敗: {stats['failed']}, "
                    f"成功率: {stats['success_rate']:.2%})"
                )
                last_report_time = current_time

        except Exception as e:
            logger.error(f"監(jiān)控進(jìn)程異常: {e}")
            time.sleep(1)  # 異常時(shí)休眠1秒

    # 最終統(tǒng)計(jì)報(bào)告
    final_stats = storage.get_statistics()
    total_elapsed_time = time.time() - start_time
    print()  # 換行
    logger.info(f"所有任務(wù)執(zhí)行完成！總耗時(shí): {total_elapsed_time:.2f} 秒, 最終統(tǒng)計(jì): {final_stats}")


def result_query_handler(storage):
    """結(jié)果查詢處理器（可擴(kuò)展為API接口）"""

    def get_results(status=None, limit=None):
        with storage.lock:
            results = list(storage.results)  # 轉(zhuǎn)換為普通列表

        if status:
            results = [r for r in results if r['status'] == status]

        if limit:
            results = results[:limit]

        return results

    return get_results


def main_enhanced():
    """增強(qiáng)的主調(diào)度函數(shù)"""
    num_tasks = 10  # 增加任務(wù)數(shù)量以更好測試異步性能
    
    # 使用Manager創(chuàng)建可以在進(jìn)程間共享的隊(duì)列和鎖
    manager = multiprocessing.Manager()
    status_queue = manager.Queue()

    # 創(chuàng)建共享存儲實(shí)例，傳入manager以便創(chuàng)建可共享的對象
    storage = AsyncResultStorage(manager=manager)

    logger.info("啟動(dòng)異步多進(jìn)程調(diào)度系統(tǒng)...")
    logger.info(f"總共 {num_tasks} 個(gè)任務(wù)需要執(zhí)行")
    start_time = time.time()

    # 啟動(dòng)監(jiān)控進(jìn)程
    monitor_process = multiprocessing.Process(
        target=monitor_status,
        args=(status_queue, num_tasks, storage)
    )
    monitor_process.start()

    # 使用ProcessPoolExecutor以獲得更好的進(jìn)程管理
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 提交所有任務(wù)
        future_to_task = {
            executor.submit(send_request, task_id, status_queue, storage): task_id
            for task_id in range(num_tasks)
        }

        # 等待所有任務(wù)完成（可選，因?yàn)楸O(jiān)控進(jìn)程會跟蹤完成狀態(tài)）
        completed_futures = 0
        for future in as_completed(future_to_task):
            task_id = future_to_task[future]
            try:
                result = future.result(timeout=30)  # 設(shè)置超時(shí)時(shí)間
                completed_futures += 1
            except Exception as e:
                logger.error(f"任務(wù) {task_id} 執(zhí)行出現(xiàn)異常: {e}")

    # 等待監(jiān)控進(jìn)程結(jié)束
    monitor_process.join(timeout=10)
    if monitor_process.is_alive():
        logger.warning("監(jiān)控進(jìn)程未正常結(jié)束，強(qiáng)制終止")
        monitor_process.terminate()

    # 計(jì)算執(zhí)行時(shí)間
    total_time = time.time() - start_time

    # 輸出最終結(jié)果和統(tǒng)計(jì)信息
    stats = storage.get_statistics()
    logger.info(f"系統(tǒng)執(zhí)行完成，總耗時(shí): {total_time:.2f} 秒")
    logger.info(f"最終統(tǒng)計(jì): 共{stats['total_tasks']}個(gè)任務(wù), "
                f"成功{stats['completed']}個(gè), 失敗{stats['failed']}個(gè), "
                f"成功率: {stats['success_rate']:.2%}")

    # 返回結(jié)果查詢接口
    return result_query_handler(storage)


if __name__ == "__main__":
    # 使用Manager創(chuàng)建可以在進(jìn)程間共享的對象
    manager = multiprocessing.Manager()
    
    # 創(chuàng)建測試存儲文件
    test_storage_file = f"async_results_{int(time.time())}.json"
    storage = AsyncResultStorage(test_storage_file, manager)

    # 運(yùn)行增強(qiáng)版系統(tǒng)
    query_handler = main_enhanced()

    # 示例：查詢執(zhí)行結(jié)果
    print("\n=== 執(zhí)行結(jié)果查詢 ===")
    all_results = query_handler()
    completed_results = query_handler(status="Completed")
    failed_results = query_handler(status="Failed")

    print(f"總?cè)蝿?wù)數(shù): {len(all_results)}")
    print(f"成功任務(wù): {len(completed_results)}")
    print(f"失敗任務(wù): {len(failed_results)}")

    # 顯示最近3個(gè)任務(wù)的結(jié)果
    print("\n最近3個(gè)任務(wù)的結(jié)果:")
    for result in all_results[-3:]:
        print(f"  任務(wù) {result['task_id']}: {result['status']} - {result['message']}")