from multiprocessing import Process
Process(target,name,args,kwargs)
'''''''''''''''''
功能 ： 创建进程对象
参数 ：  
	  target 绑定要执行的目标函数 
   	  name 进程名，默认是Process-x(整数)
	  args 元组，用于给target函数位置传参
	  kwargs 字典，给target函数键值传参
'''''''''''''''''''

from multiprocessing import Process
import time
import os


# 创建进程的标准方法
def worker_function(name, age):
    """子进程执行过程中触发的函数"""
    print(f"子进程ID:{os.getpid()},父进程ID{os.getppid()}")
    print(f"子进程正在执行，参数name={name},age={age}")
    # 模拟耗时操作
    time.sleep(5)
    print(f"子进程{name}执行完毕")


def cpu_intensive_task(number):
    """CPU密集型任务"""
    result = 0
    for i in range(number):
        result += i * i
    print(f"进程{os.getpid()} 计算完成，结果为{result}")


if __name__ == '__main__':
    print(f"父进程ID:{os.getpid()}")
    start_time = time.time()

    # 创建多个进程，体现并行处理能力
    processes = []

    # 创建四个检测执行不同人物
    p1 = Process(target=worker_function, args=("张三",), kwargs={"age": 20})
    p2 = Process(target=worker_function, args=("李四",), kwargs={"age": 30})
    p3 = Process(target=cpu_intensive_task, args=(10000,))
    p4 = Process(target=cpu_intensive_task, args=(20000,))

    processes.extend([p1, p2, p3, p4]) # 将进程添加到列表中

    for p in processes:
        p.start() # 启动进程

    # 等待所有进程结束
    for p in processes:
        p.join()

    end_time = time.time()
    print(f"所有进程执行完毕，总耗时{end_time - start_time:.2f}秒")
    print("如果使用单进程顺序执行，耗时会更长，因为是两个任务在执行，多进程可以充分利用多核CPU并行处理任务")

from multiprocessing import Process
import time
import os


# 通过继承Process类创建进程
class WorkerProcess(Process):
    """继承Process类的自定义进程"""
    def __init__(self, name, age=None):
        super().__init__()
        self.name = name
        self.age = age
    
    def run(self):
        """重写run方法，进程启动后会执行该方法"""
        print(f"子进程ID:{os.getpid()},父进程ID{os.getppid()}")
        print(f"子进程正在执行，参数name={self.name},age={self.age}")
        # 模拟耗时操作
        time.sleep(5)
        print(f"子进程{self.name}执行完毕")


class CPUIntensiveProcess(Process):
    """CPU密集型任务进程"""
    def __init__(self, number):
        super().__init__()
        self.number = number
    
    def run(self):
        """重写run方法，执行CPU密集型计算"""
        result = 0
        for i in range(self.number):
            result += i * i
        print(f"进程{os.getpid()} 计算完成，结果为{result}")


if __name__ == '__main__':
    print(f"父进程ID:{os.getpid()}")
    start_time = time.time()

    # 创建多个进程，体现并行处理能力
    processes = []

    # 创建四个进程实例
    p1 = WorkerProcess("张三", 20)
    p2 = WorkerProcess("李四", 30)
    p3 = CPUIntensiveProcess(10000)
    p4 = CPUIntensiveProcess(20000)

    processes.extend([p1, p2, p3, p4]) # 将进程添加到列表中

    for p in processes:
        p.start() # 启动进程

    # 等待所有进程结束
    for p in processes:
        p.join()

    end_time = time.time()
    print(f"所有进程执行完毕，总耗时{end_time - start_time:.2f}秒")

import multiprocessing
import os

def process_info():
    """打印当前进程的信息"""
    # 获取当前进程对象
    process = multiprocessing.current_process()
    
    print(f"进程名称: {process.name}")
    print(f"进程ID: {process.pid}")
    print(f"父进程ID: {os.getppid()}")
    print(f"进程授权键: {process.authkey}")
    print(f"进程是否活跃: {process.is_alive()}")
    
if __name__ == '__main__':
    p = multiprocessing.Process(target=process_info, name="自定义进程名")
    p.start()
    p.join()

from multiprocessing import Process,Queue
import time

## 子进程执行的函数
def worker(q):
    """子进程函数，想往父进程发送消息，就往这个队列里放"""
    print("子进程启动")
    # 向队列中添加数据
    q.put("我是一个队列数据")
    time.sleep(2) # 模拟任务执行
    print("子进程结束")

if __name__ == '__main__':
    q = Queue() # 父进程创建队列
    # 创建一个子进程对象
    p = Process(target=worker, args=(q,))
    # 启动子进程
    p.start()
    # 主进程从队列中获取数据
    print("主进程等待子进程数据....")
    message = q.get() # 阻塞等待数据
    print(f"主进程收到来自于子进程的消息：{message}")
    # 等待子进程结束
    p.join()
    print("主进程结束")

import os
import time
from multiprocessing import Process, Queue, Lock, Value, Array


## 子进程执行的函数
def worker_with_args(q, lock, value, arr, sleep_num):
    """带有共享资源的工作函数
    Args:
        q: 队列
        lock: 锁
        value: 值
        arr: 数组
        sleep_num: 休眠时间
    """
    # 获取进程id
    pid = os.getpid()
    print(f"进程{pid}开始执行")

    # 使用锁来保证数据安全
    with lock:  # 相当于lock.acquire()和lock.release()的组合
        value.value += 1
        for i in range(len(arr)):
            arr[i] **= 2  # 安全的修改数组元素
            print(f"进程{pid}修改数组元素{i}为{arr[i]}")
    # 向队列中添加数据
    q.put(f"这是一条来自于进程{pid}的信息")

    # 休眠模拟工作
    time.sleep(sleep_num)

    print(f"进程{pid}执行完毕")


if __name__ == "__main__":
    """
    解释输出逻辑:
    1. 创建了4个进程，它们共享同一个数组arr，初始值为[1,2,3,4,5]
    2. 每个进程获取锁后，会将数组中的每个元素进行平方操作(arr[i] **= 2)
    3. 由于进程是按顺序启动的，但执行顺序不确定，所以:
       - 第一个获得锁的进程将[1,2,3,4,5]平方为[1,4,9,16,25]
       - 第二个获得锁的进程将[1,4,9,16,25]平方为[1,16,81,256,625]
       - 第三个获得锁的进程将[1,16,81,256,625]平方为[1,256,6561,65536,390625]
       - 第四个获得锁的进程将[1,256,6561,65536,390625]平方，但由于整数溢出，
         导致最后两个元素变成了0和负数
    4. 进程完成的顺序取决于sleep_num参数(1,2,3,4)，所以最先完成的是第一个进程
    """
    # 创建队列
    q = Queue()
    # 创建锁
    lock = Lock()
    # 创建一个共享值
    value = Value("i", 0)  # "i"表示int类型

    # 创建一个共享数组
    arr = Array("i", [1, 2, 3, 4, 5])

    # 创建多个子进程
    processes = []
    for i in range(4):
        p = Process(target=worker_with_args, args=(q, lock, value, arr, i + 1))
        p.start()
        processes.extend([p])

from concurrent.futures import ProcessPoolExecutor
import os

def worker_function(name,age):
    """进程池工作函数"""
    pid = os.getpid()
    print(f"进程{pid}：{name}，{age}岁")
    return f"我的父进程是{os.getppid()} 我结束进程了"

if __name__ == '__main__':
    with ProcessPoolExecutor(max_workers=4) as executor:
        for i in range(10):
            # submit方法提交任务到进程池，返回一个Future对象
            # submit参数: function, *args, **kwargs
            # 这里的i作为worker_function函数的第二个参数age传入
            future = executor.submit(worker_function, f"小明{i}", i)
            # 等待future对象返回结果
            result = future.result()
            print(result)

from multiprocessing import Pool
import os

def worker_function(args):
    """工作进程函数"""
    name,age = args
    pid = os.getpid()
    print(f"进程{pid}：{name}，{age}岁")
    return f"我的父进程是{os.getppid()} 我结束进程了"

if __name__ == '__main__':
    with Pool() as pool:
        # 准备参数列表
        args_list = [(f"张三{i}号", i+18) for i in range(1,10)]
        
        # map方法将函数应用于参数列表，并返回结果列表
        # pool.map 的工作原理:
        # 1. 它接收两个参数：要执行的函数(worker_function)和可迭代的参数列表(args_list)
        # 2. 它会自动将参数列表中的每个元素分配给不同的进程来执行
        # 3. 每个进程会调用worker_function并传入args_list中的一个元素作为参数
        # 4. 所有进程执行完毕后，map会收集所有进程的返回值，并按原始参数的顺序返回结果列表
        # 5. 这样实现了并行处理，提高了计算效率
        result_list = pool.map(worker_function, args_list)
        
        # 打印每个进程的返回结果
        for result in result_list:
            print(result)

import multiprocessing
import os

def process_info():
    """打印当前进程的信息"""
    # 获取当前进程对象
    process = multiprocessing.current_process()
    
    print(f"进程名称: {process.name}")
    print(f"进程ID: {process.pid}")
    print(f"父进程ID: {os.getppid()}")
    print(f"进程授权键: {process.authkey}") # 授权键用于在进程间通信时进行身份验证。
    print(f"进程是否活跃: {process.is_alive()}")
    
if __name__ == '__main__':
    p = multiprocessing.Process(target=process_info, name="自定义进程名")
    p.start()
    p.join()

'''
子进程死后还会有一些资源占用(进程号，进程的运行状态，运行时间)，等待父进程通过系统调用
进行资源回收

相当于子进程死了之后，需要父进程来给他"收尸"
除了init进程之外，所有的进程最后都会步入僵尸进程
在一种情况下是会带来危害的:
子进程退出之后，父进程没有及时处理，僵尸进程就会一直占用资源
如果产生了大量僵尸进程，资源过度使用，系统没有可用的进程号，导致系统不能产生新的进程
'''

'''
子进程处于存活状态，父进程意外死亡，操作系统就会开设一个孤儿院（init进程），用来管理
孤儿进程，回收孤儿进程相关资源
'''

from threading import Thread
import time

def worker_function(name,delay):
    print(f"线程{name}开始工作")
    time.sleep(delay)
    print(f"线程{name}结束工作")

if __name__ == '__main__':
    t1 = Thread(target=worker_function,args=("线程1",2))
    t2 = Thread(target=worker_function,args=("线程2",4))
    # 启动线程
    t1.start()
    t2.start()
    # 等待线程结束
    t1.join()
    t2.join()
    print("主线程结束")

from threading import Thread
import time
class MyThread(Thread):
    def __init__(self,name,message):
        super().__init__()
        self.name = name
        self.message = message

    def run(self):
        print(f"线程{self.name}开始执行")
        time.sleep(2)
        print(f"线程{self.name}执行完毕，消息：{self.message}")

if __name__ == '__main__':
    t1 = MyThread(name="线程1",message="子进程操作完毕")
    t2 = MyThread(name="线程2",message="子进程操作完毕")
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("主进程执行完毕")

from threading import Thread
import time
import requests

def download_file(url,session):
    """访问网站下载文件"""
    with session.get(url) as response:
        print(f"读取{url} 长度为{len(response.content)}")


def download_all_sites(urls):
    """单线程下载所有网站"""
    with requests.Session() as session:
        for url in urls:
            download_file(url,session)
            time.sleep(1)

def download_site_thread(url,session):
    """多线程下载网站"""
    download_file(url,session)

def download_all_sites_thread(urls):
    """多线程下载所有网站"""
    threads = []
    with requests.Session() as session:
        for url in urls:
            thread = Thread(target=download_site_thread, args=(url,session))
            threads.append(thread)
            thread.start()
        # 等待所有线程结束
        for thread in threads:
            thread.join()
if __name__ == '__main__':
    # 准备一些网站用于演示
    sites = [
        "https://www.baidu.com",
        "https://www.sina.com.cn",
        "https://www.qq.com",
        "https://www.163.com",
        "https://www.sohu.com",
    ]
    # 单线程下载
    print("=======单线程下载开始========")
    start_time = time.time()
    download_all_sites(sites)
    end_time = time.time()
    print(f"单线程下载结束，耗时{end_time-start_time}秒")

    print("\n=======多线程下载开始========")
    start_time = time.time()
    download_all_sites_thread(sites)
    end_time = time.time()
    print(f"多线程下载结束，耗时{end_time-start_time}秒")
    print("\nIO密集型任务（如网络请求）适合使用多线程，可以显著提高性能")
    print("这是因为当一个线程等待IO操作完成时，其他线程可以继续执行")

from threading import Thread
import time

def worker_function(thread_name):
    print(f"线程 {thread_name} 启动")
    time.sleep(3)
    print(f"线程 {thread_name} 结束")


if __name__ == '__main__':
    t1 = Thread(target=worker_function, args=("Thread-1",))
    t2 = Thread(target=worker_function, args=("Thread-2",))
    # 设置t3为守护线程，主线程结束时，t3线程也会结束
    t3 = Thread(target=worker_function, args=("Thread-3",), daemon=True)
    # t3.setDaemon(True) # 已经被废弃的API，现在使用daemon=True参数代替
    # 启动普通线程
    t1.start()
    t2.start()
    # 启动守护线程
    t3.start()  # 这里需要启动t3线程，否则t3不会执行

    # 等待普通线程完成
    print(f"等待线程{t1.name} + {t2.name}完成...")
    t1.join()
    t2.join()

    # 检测线程状态
    print(f"线程{t1.name}是否存活：{t1.is_alive()}")
    print(f"线程{t2.name}是否存活：{t2.is_alive()}")
    print(f"线程{t3.name}是否存活：{t3.is_alive()}")  # True

    # 主线程休眠一段时间，以便守护线程有机会执行
    time.sleep(10)
    print("主线程结束")

from threading import Thread
import time
import logging
import psutil
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s',
)

def system_monitor(interval=1):
    """
    守护线程：系统资源监控器
    持续监控CPU使用率和内存使用情况，并记录到日志中
    """
    logging.info('系统监控守护线程启动')
    try:
        while True:
            cpu_usage = psutil.cpu_percent(interval=interval)
            mem_usage = psutil.virtual_memory().percent
            logging.info(f'CPU使用率：{cpu_usage}% 内存使用率：{mem_usage}%')
            time.sleep(interval)
            if cpu_usage > 80 or mem_usage > 80:
                logging.warning('系统资源占用过高，请及时处理')
    except Exception as e:
        logging.error(f'系统监控线程异常：{e}')
    finally:
        logging.info('系统监控守护线程结束')
if __name__ == '__main__':
    # 创建并启动系统监控守护线程
    monitor_thread = Thread(target=system_monitor,args=(1,),daemon=True,name="MonitorThread")
    monitor_thread.start()
    # 主线程继续执行一段时间，守护线程在后台运行
    logging.info("主线程运行中，监控守护线程在后台运行...")
    time.sleep(30)  # 运行30秒后结束
    # 主线程结束，守护线程将自动终止
    logging.info(f"监控守护线程是否存活: {monitor_thread.is_alive()}")
    logging.info("主线程结束，守护线程将自动终止")

from concurrent.futures import ThreadPoolExecutor
import time


# 定义一个耗时函数
def time_consuming_task(n):
    time.sleep(1)
    return n * n


if __name__ == '__main__':
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=4) as executor:
        # 使用submit方法提交任务给线程池执行，返回Future对象列表
        futures = [executor.submit(time_consuming_task, i) for i in range(1, 21)]
        print("任务已提交，主线程继续执行.........")
        # 等待所有任务完成，并获取结果
        results = [future.result() for future in futures]
        print(f"任务执行完毕，结果为：{results}")
    end_time = time.time()
    print(f"总共耗时：{end_time - start_time}秒")
    ### 线程池在with语句结束时自动关闭

from concurrent.futures import ThreadPoolExecutor,as_completed,wait
import time

def time_consuming_task(n):
    """耗时任务"""
    time.sleep(n % 3 +1) # 不同的n值，耗时不同
    return n * n

def task_done_callback(future):
    """任务完成后触发的回调函数"""
    print(f"任务完成，结果为{future.result()}")

if __name__ == '__main__':
    print("===== 1. submit方法示例 =====")
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=3) as executor:
        future_list = [executor.submit(time_consuming_task, i) for i in range(10)]
        print("提交任务完成，等待结果...")
        # result()：获取任务执行结果，会阻塞直到所有任务完成
        # as_completed(): 返回已完成任务的迭代器
        for future in as_completed(future_list):
            print(f"任务{future.result()}完成")
    end_time = time.time()
    print(f"总耗时：{end_time - start_time}秒")

    print("\n===== 2. map方法示例 =====")
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=4) as executor:
        # map()：对每一个输入并行执行函数，返回结果迭代器
        # 与submit不同，map会自动收集结果并按输入顺序返回
        # 不需要手动调用future.result()
        results = executor.map(time_consuming_task, range(10))
        print("任务提交完成，直接获取有序结果...")
        # 转换为列表时会按照输入顺序返回结果，如果任务未完成会在这里阻塞等待
        results_list = list(results)
        print(f"结果：{results_list}")
    end_time = time.time()
    print(f"总耗时：{end_time - start_time}秒")

    print("\n===== 3. add_done_callback示例 =====")
    # add_done_callback(fn): 添加任务完成回调函数
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        for i in range(1, 6):
            future = executor.submit(time_consuming_task, i)
            # 添加回调函数，任务完成后自动调用
            future.add_done_callback(task_done_callback)
            futures.append(future)
        print("已添加回调函数，主线程继续执行...")
        # 等待所有任务完成
        for future in futures:
            future.result()
    print(f"耗时：{time.time() - start_time}秒")
    # 适用场景：任务完成后的后续处理，适合需要在任务完成时执行额外操作而不阻塞主线程

    print("\n===== 4. wait示例 =====")
    # wait(): 等待任务完成
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(time_consuming_task, i) for i in range(1, 11)]
        print("任务已提交，等待所有任务完成...")
        # 等待所有任务完成
        done, not_done = wait(futures)
        print(f"完成的任务数: {len(done)}, 未完成的任务数: {len(not_done)}")
        results = [future.result() for future in done]
        print(f"所有任务执行完毕，结果为：{results}")
    print(f"耗时：{time.time() - start_time}秒")
    # 适用场景：任务同步点，适合需要等待一组任务全部或部分完成后再继续执行的情况

from threading import Thread, Event
import time

## 创建一个事件对象
event = Event()

## 模拟公交车到站的函数
def bus_stop():
    """模拟公交车到站过程"""
    print('公交车即将到站')
    time.sleep(2)  # 模拟行驶时间
    print('公交车已到站<====>'*10)
    # 设置事件，通知等待的乘客
    event.set()  # 发射信号，让等车的人上车


def passenger(name):
    """模拟乘客等车
    
    Args:
        name: 乘客名称
    """
    # 等待公交车到站
    print(name, '等车中')
    event.wait()  # 阻塞等待信号
    print(name, '出发！！！！！！！！！！！！！！！！！！！！！！！！！！')


if __name__ == '__main__':
    # 创建公交车线程
    t1 = Thread(target=bus_stop)
    t1.start()

    # 创建多个乘客线程
    for i in range(20):
        t = Thread(target=passenger, args=(f'乘客{i}',))
        t.start()
        time.sleep(0.1)  # 模拟乘客陆续到站

from threading import Timer

def delayed_greeting(name):
    """延迟执行的问候函数
    
    Args:
        name: 要问候的对象名称
    """
    print(f"{name}说: 哈哈，我是延迟1秒后才执行的!")

## 创建一个定时器，1秒后执行hello函数，参数为"小明"
timer = Timer(1, delayed_greeting, args=("小明",))
## 启动定时器
timer.start()

print("定时器已启动，但greeting函数还未执行...")
## 主线程继续执行，不会被阻塞

from multiprocessing import Process
from threading import Thread
import time

'''
计算密集型任务对比测试
'''
def task():
    """计算密集型任务"""
    res = 0
    for i in range(10000000):  # 执行大量计算
        res += i

if __name__ == '__main__':
    start_time = time.time()
    l = []

    for i in range(10):
        # 使用多进程或多线程(取消相应的注释来测试)
        p = Process(target=task)  # 多进程：结果大概是1.65秒
        # p = Thread(target=task)   # 多线程：结果大概是4.18秒
        
        p.start()
        l.append(p)

    for p in l:
        p.join()

    end = time.time()
    print("花费时间", end - start_time)

from multiprocessing import Process
from threading import Thread
import time

'''
IO密集型任务对比测试
'''
def task():
    """IO密集型任务，使用sleep模拟IO操作"""
    time.sleep(1)  # 模拟IO等待
    
if __name__ == '__main__':
    start_time = time.time()
    l = []

    for i in range(100):  # 创建100个任务
        # 使用多进程或多线程(取消相应的注释来测试)
        # p = Process(target=task)  # 多进程：结果约19.34秒
        p = Thread(target=task)   # 多线程：结果约1.01秒
        
        p.start()
        l.append(p)

    for p in l:
        p.join()

    end = time.time()
    print("花费时间", end - start_time)

'''
进程：资源单位 - 系统分配资源的基本单位，拥有独立的内存空间
线程：执行单位 - CPU调度和执行的最小单位，共享所属进程的内存空间
协程：根本不存在，它是程序员人为创造出来的(切换+保存状态)
当程序遇到IO的时候，通过我们的代码，让我们的程序自动完成切换
也就是通过代码监听IO，一旦程序遇到IO，就在代码层面自动切换，给CPU的感觉就是我们的程序没有IO
换句话说也就是我们欺骗了CPU
'''

import time

def f1():
    """计算密集型函数1"""
    n = 0
    for i in range(10000000):
        n += i  # 执行简单累加计算

def f2():
    """计算密集型函数2"""
    n = 0
    for i in range(10000000):
        n += i  # 执行简单累加计算

start_time = time.time()
f1()  # 顺序执行f1
f2()  # 然后执行f2
## 保留两位小数
print("串行执行总共用时：%.2f秒" % (time.time() - start_time))  # 串行执行总共用时：0.84秒

import time

def f1():
    """带yield的计算密集型函数1"""
    n = 0
    for i in range(10000000):
        n += i
        yield  # 主动让出执行权，保存当前执行状态

def f2():
    """使用f1的生成器进行交替执行"""
    g = f1()  # 创建生成器对象
    n = 0
    for i in range(10000000):
        n += i
        next(g)  # 切换到f1执行一步，f1会执行到下一个yield后暂停

start_time = time.time()
f2()  # 执行f2，内部会与f1交替执行
## 保留两位小数
print("yield协程用时：%.2f秒" % (time.time() - start_time))  # 约1.45秒

from greenlet import greenlet
import time

def func_a():
    """协程函数a"""
    while True:
        print('函数a正在运行')
        time.sleep(1)  # 模拟某些操作
        b.switch()  # 主动切换到函数b执行

def func_b():
    """协程函数b"""
    while True:
        print('函数b正在运行')
        time.sleep(2)  # 模拟某些操作
        a.switch()  # 切换回函数a执行

if __name__ == '__main__':
    # 创建两个greenlet对象
    a = greenlet(func_a)  # 将函数封装为greenlet对象
    b = greenlet(func_b)  # 将函数封装为greenlet对象
    # 从函数a开始执行
    a.switch()  # 启动协程a

'''
gevent 是一个基于协程的 Python 网络库，它使用 greenlet 在 libev 或 libuv 
等事件循环之上提供高级同步 API。gevent 实现了python 标准库里面大部分的阻塞式系统调用，
包括 socket、ssl、threading 和 select 等模块，
可以使用 "猴子补丁" 将这些阻塞式调用变为协作式运行。

猴子补丁的功能很强大，但是也带来了很多的风险，尤其是像 gevent 这种直接进行 API替换的补丁，
整个 Python 进程所使用的模块都会被替换，可能自己的代码能 hold 住，
但是其它第三方库，有时候问题并不好排查，即使排查出来也是很棘手，所以，
就像松本建议的那样，如果要使用猴子补丁，那么只是做功能追加，
尽量避免大规模的 API 覆盖。 虽然猴子补丁仍然是邪恶的(evil)，
但在这种情况下它是 "有用的邪恶(useful evil)"。
'''

import gevent
from gevent import monkey

## 应用猴子补丁，将标准库的阻塞操作替换为非阻塞版本
## 必须在导入其他模块前调用，确保所有IO操作都被替换
monkey.patch_all()  # 替换所有可能的阻塞调用

def foo():
    """协程函数1"""
    print('Running in foo')
    gevent.sleep(0)  # 模拟IO操作，主动让出控制权
    print('Explicit context switch to foo')

def bar():
    """协程函数2"""
    print('Running in bar')
    gevent.sleep(0)  # 模拟IO操作，主动让出控制权
    print('Explicit context switch to bar')

def baz():
    """协程函数3"""
    print('Running in baz')
    gevent.sleep(0)  # 模拟IO操作，主动让出控制权
    print('Explicit context switch to baz')

## 创建三个协程
g1 = gevent.spawn(foo)  # 创建协程但不立即执行
g2 = gevent.spawn(bar)  # 创建协程但不立即执行
g3 = gevent.spawn(baz)  # 创建协程但不立即执行

## 等待所有协程完成
gevent.joinall([g1, g2, g3])  # 类似于多线程中的join方法

import gevent
from gevent import monkey
import requests
import time

## 替换标准库
monkey.patch_all()

def fetch_url(url):
    """获取URL内容的函数
    
    Args:
        url: 要获取的网址
        
    Returns:
        tuple: (url, 响应状态码, 内容长度)
    """
    try:
        print(f"开始请求: {url}")
        start = time.time()
        response = requests.get(url, timeout=5)  # 进行HTTP请求，IO操作会自动切换
        elapsed = time.time() - start
        print(f"完成请求: {url}, 耗时: {elapsed:.2f}秒")
        return url, response.status_code, len(response.content)
    except Exception as e:
        print(f"请求 {url} 出错: {e}")
        return url, 0, 0

## 要获取的URL列表
urls = [
    "https://www.python.org",
    "https://www.github.com",
    "https://www.stackoverflow.com",
    "https://www.wikipedia.org",
    "https://www.reddit.com"
]

start_time = time.time()

## 创建协程任务
tasks = [gevent.spawn(fetch_url, url) for url in urls]

## 等待所有任务完成
gevent.joinall(tasks)

## 获取结果
results = [task.value for task in tasks]

## 打印结果
print("\n结果汇总:")
for url, status, length in results:
    print(f"URL: {url}, 状态码: {status}, 内容长度: {length} 字节")

print(f"\n总耗时: {time.time() - start_time:.2f}秒")

import asyncio
import time

async def fetch_data(url,delay):
    """模拟从网络获取数据的异步函数"""
    print(f"开始获取数据：{url}，延迟{delay}秒")
    await asyncio.sleep(delay)
    print(f"成功获取数据长度：{len(url)}")
    return f"数据{url}"

async def main():
    """异步操作的主函数"""
    print(f"程序开始时间:{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")

    print("\n===== 串行执行示例 =====")
    start_time = time.time()
    # 串行执行 - 请求两个API数据
    result1 = await fetch_data("https://www.baidu.com", 2)
    result2 = await fetch_data("https://www.sina.com.cn",3)
    print(f"串行执行结果：{result1}, {result2}")
    end_time = time.time()
    print(f"程序结束时间:{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")
    print(f"程序耗时:{end_time-start_time}秒")

    # 并行执行 - 请求两个API数据
    print("\n===== 并行执行示例 =====")
    start_time = time.time()
    tasks = [
        asyncio.create_task(fetch_data("https://www.baidu.com", 2)),
        asyncio.create_task(fetch_data("https://www.sina.com.cn",3)),
    ]
    results = await asyncio.gather(*tasks) # 批量等待所有任务完成
    print(f"并行执行结果：{results}")
    end_time = time.time()
    print(f"程序结束时间:{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")
    print(f"程序耗时:{end_time-start_time}秒")

if __name__ == '__main__':
    # 在Python 3.7+中，可以直接使用asyncio.run()运行主协程
    asyncio.run(main())

import asyncio

async def nested():
    print('进入 nested()')
    await asyncio.sleep(1)  # 模拟IO操作
    print('离开 nested()')
    return '42'

async def main():
    task = asyncio.create_task(nested())  # 创建任务
    result = await task  # 等待任务完成
    print(f'返回值：{result}')

asyncio.run(main())

import asyncio

# 定义一个异步函数，用于设置future的结果
async def set_future_result(future):
    # 异步等待2秒，模拟耗时操作
    await asyncio.sleep(2)
    # 设置future的结果为"Hello, world!"
    future.set_result("Hello, world!")

# 定义主异步函数
async def main():
    # 获取当前正在运行的事件循环
    loop = asyncio.get_running_loop()
    # 创建Future对象，它代表一个尚未完成的异步操作
    future = loop.create_future()  # 创建Future对象
    # 创建一个任务来执行set_future_result函数，不等待其完成立即返回
    asyncio.create_task(set_future_result(future))
    # 等待future完成并获取其结果
    result = await future  # 等待Future完成
    # 打印future的结果
    print(f"Future的结果: {result}")

# 运行主异步函数
asyncio.run(main())

import asyncio

class AsyncResource:
    async def __aenter__(self):
        print("资源获取")
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        print("资源释放")

async def main():
    async with AsyncResource():
        print("执行任务中")

asyncio.run(main())

'''
GIL全局解释器锁(Global Interpreter Lock)，是CPython特有的一个物件，
作用是让一个进程中同一时刻只能有一个线程可以被CPU调用

如果程序想利用计算机的多核优势，让CPU同时处理一些任务，适合用多进程开发（即使资源开销大）
如果程序不想利用计算机的多核优势，适合用多线程开发
'''

## GIL工作示意伪代码
def thread_execution():
    while True:
        acquire_GIL()       # 获取GIL锁
        execute_bytecodes() # 执行一定数量的字节码
        release_GIL()       # 释放GIL锁以允许其他线程运行
        wait_for_GIL()      # 等待再次获取GIL
        
# 这也就导致了每一个线程都需要在执行获取字节码时都要经历拿锁->解锁的过程

from threading import Thread
import time
import random

### 共享的全局变量
counter = 0
start_time = time.time()
iterations_completed = 0

def increment_counter():
    """增加计数器，但使用了非原子操作的方式"""
    global counter, iterations_completed
    for _ in range(1000000):
        # 模拟线程安全问题：读取-修改-写入过程中可能被中断
        local_counter = counter  # 读取当前值
        
        # 模拟线程在读取后被切换的情况
        # 随机休眠一个很小的时间，增加线程切换的可能性
        if random.random() < 0.00001:
            time.sleep(0.00001)
            
        local_counter += 1  # 在本地修改
        counter = local_counter  # 写回全局变量
        iterations_completed += 1


def run_concurrent_threads(num_threads):
    """运行多个线程同时增加计数器"""
    global counter, iterations_completed
    counter = 0
    iterations_completed = 0
    
    threads = []
    for _ in range(num_threads):
        t = Thread(target=increment_counter)
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    # 理论上应该等于 num_threads * 1000000
    expected = num_threads * 1000000
    print(f"预期结果: {expected}")
    print(f"实际结果: {counter}")
    print(f"丢失的增量: {expected - counter}")
    print(f"完成的迭代次数: {iterations_completed}")


if __name__ == '__main__':
    print(f'开始时间为: {time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}')
    
    # 运行4个线程，每个线程增加计数器1000000次
    # 理论上最终结果应该是4000000，但由于线程安全问题，实际结果会小于这个值
    run_concurrent_threads(4)
    
    print(f"累计用时: {round(time.time() - start_time, 1)}秒")
    

from threading import Thread, Lock
import time
import random

### 共享的全局变量
counter = 0
start_time = time.time()
iterations_completed = 0
# 创建一个线程锁
counter_lock = Lock()


def increment_counter():
    """增加计数器，使用线程锁确保线程安全"""
    global counter, iterations_completed
    for _ in range(1000000):
        # 使用线程锁保护临界区
        with counter_lock:
            counter += 1  # 在锁的保护下直接修改全局变量
            iterations_completed += 1


def run_concurrent_threads(num_threads):
    """运行多个线程同时增加计数器"""
    global counter, iterations_completed
    counter = 0
    iterations_completed = 0

    threads = []
    for _ in range(num_threads):
        t = Thread(target=increment_counter)
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    # 理论上应该等于 num_threads * 1000000
    expected = num_threads * 1000000
    print(f"预期结果: {expected}")
    print(f"实际结果: {counter}")
    print(f"丢失的增量: {expected - counter}")
    print(f"完成的迭代次数: {iterations_completed}")


if __name__ == '__main__':
    print(f'开始时间为: {time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}')

    # 运行4个线程，每个线程增加计数器1000000次
    # 使用线程锁后，最终结果应该正确等于4000000
    run_concurrent_threads(4)

    print(f"累计用时: {round(time.time() - start_time, 1)}秒")

import concurrent
import threading
import time
from concurrent.futures import ThreadPoolExecutor

lock = threading.Lock()
shared_data = 0


def shared_resource(thread_id: int):
    """访问共享资源的函数
    Args:thread_id: 线程ID，用于标识不同线程
    """
    # 尝试获取锁
    if lock.acquire(timeout=1):  # 添加超时参数，防止无限等待
        print(f"线程{thread_id}获取锁")
        try:
            global shared_data
            # 读取-修改-写入操作需要原子性保护
            current = shared_data
            time.sleep(0.1)  # 模拟处理延时，增加竞争概率
            shared_data = current + 1
            print(f"线程{thread_id}修改共享数据，当前值为{shared_data}")
        finally:
            # 释放锁
            lock.release()
            print(f"线程{thread_id}释放锁")
    else:
        print(f"线程{thread_id}获取锁失败")
        

def shared_resource2(thread_id: int):
    """访问共享资源的函数
    使用with语句，自动释放锁简化代码
    """
    with lock:
        print(f"线程{thread_id}获取锁")
        try:
            global shared_data
            # 读取-修改-写入操作需要原子性保护
            current = shared_data
            time.sleep(0.1)  # 模拟处理延时，增加竞争概率
            shared_data = current + 1
            print(f"线程{thread_id}修改共享数据，当前值为{shared_data}")
        finally:
            print(f"线程{thread_id}释放锁")


if __name__ == '__main__':
    # 使用线程池创建多个线程
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(shared_resource, i) for i in range(1, 11)]
        # 等待所有线程完成
        concurrent.futures.wait(futures)
    print(f"最终共享数据值为{shared_data}")

import threading
import time

## 创建可重入锁
rlock = threading.RLock()

# 嵌套列表数据结构
data = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]


def process_data(item, depth: int = 0):
    """递归处理数据
    
    Args:
        item: 要处理的数据项，可以是列表或单个元素
        depth: 当前递归深度，用于缩进显示
    """
    # 获取锁
    rlock.acquire()
    try:
        # 创建缩进效果，增强可读性
        indent = " " * depth * 2  # 增加缩进量使层次更明显
        
        # 打印当前处理的数据项和深度
        print(f'{indent}线程 {threading.current_thread().name} 处理: {item} (深度: {depth})')
        
        # 递归处理逻辑
        if isinstance(item, list):
            # 列表节点处理 - 继续向下递归
            print(f'{indent}├── 发现列表，开始遍历子元素...')
            for i, sub_item in enumerate(item):
                # 显示子项的索引，增强结构可视化
                prefix = "└── " if i == len(item) - 1 else "├── "
                print(f'{indent}{prefix}处理子项 {i+1}/{len(item)}: {sub_item}')
                
                # 递归调用，这里会再次获取同一个锁
                process_data(sub_item, depth + 1)
                time.sleep(0.1)
        else:
            # 叶子节点处理 - 递归终止条件
            print(f'{indent}└── 发现元素，进行处理...')
            time.sleep(0.5)
            print(f'{indent}    处理结果: {item * 2}')
    finally:
        # 释放锁
        rlock.release()


if __name__ == '__main__':
    ## 创建多个线程访问嵌套数据
    threads = []
    for i in range(3):
        # 每个线程处理完整的数据结构
        t = threading.Thread(name=f"Thread-{i}", target=process_data, args=(data,))
        threads.append(t)
        t.start()
        time.sleep(0.5)  # 错开线程启动时间
    
    ## 等待所有线程结束
    for t in threads:
        t.join()
    
    print("所有线程都结束了")

import threading
import time
import random
from typing import List, Any


class Buffer:
    """线程安全的缓冲区，使用条件变量控制生产者消费者模型"""

    def __init__(self, max_size: int = 5) -> None:
        """初始化缓冲区"""
        self.buffer: List[Any] = []  # 共享数据缓冲区
        self.max_size: int = max_size  # 最大容量
        # 创建条件变量，基于RLock
        self.condition: threading.Condition = threading.Condition()

    def produce(self, item: Any, producer_id: int) -> None:
        """生产者方法，向缓冲区添加数据"""
        # 使用条件变量的with语句自动获取和释放锁
        with self.condition:
            # 当缓冲区已满时，等待消费者处理
            while len(self.buffer) >= self.max_size:
                print(f"生产者 {producer_id}: 缓冲区已满，等待消费者...")
                # 等待唤醒通知，自动释放锁，让其他线程能访问缓冲区
                self.condition.wait()

            # 添加数据到缓冲区
            self.buffer.append(item)
            print(f"生产者 {producer_id}: 添加 {item} 到缓冲区，当前大小: {len(self.buffer)}")

            # 通知所有等待的消费者有新数据可用
            self.condition.notify_all()

    def consume(self, consumer_id: int) -> Any:
        """消费者方法，从缓冲区获取数据"""
        with self.condition:
            # 当缓冲区为空时，等待生产者添加数据
            while len(self.buffer) == 0:
                print(f"消费者 {consumer_id}: 缓冲区为空，等待生产者...")
                self.condition.wait()

            # 从缓冲区取出数据
            item = self.buffer.pop(0)
            print(f"消费者 {consumer_id}: 从缓冲区取出 {item}，当前大小: {len(self.buffer)}")

            # 通知所有等待的生产者缓冲区有空间
            self.condition.notify_all()

            return item


def producer_task(buffer: Buffer, producer_id: int) -> None:
    """生产者任务"""
    for i in range(10):  # 生产10个产品
        item = f"产品-{producer_id}-{i}"
        # 模拟生产时间
        time.sleep(random.uniform(0.1, 0.5))  
        buffer.produce(item, producer_id)


def consumer_task(buffer: Buffer, consumer_id: int) -> None:
    """消费者任务"""
    for _ in range(7):  # 每个消费者消费7个产品
        # 模拟消费时间
        time.sleep(random.uniform(0.2, 0.7))  
        item = buffer.consume(consumer_id)
        print(f"消费者 {consumer_id} 处理 {item}")


def main() -> None:
    """主函数，创建并启动生产者和消费者线程"""
    # 创建共享缓冲区
    shared_buffer = Buffer(max_size=3)
    
    # 创建生产者和消费者线程
    producer_threads = [
        threading.Thread(target=producer_task, args=(shared_buffer, i), name=f"Producer-{i}")
        for i in range(3)  # 3个生产者
    ]
    
    consumer_threads = [
        threading.Thread(target=consumer_task, args=(shared_buffer, i), name=f"Consumer-{i}")
        for i in range(3)  # 3个消费者
    ]
    
    # 启动所有线程
    all_threads = producer_threads + consumer_threads
    for thread in all_threads:
        thread.start()
    
    # 等待所有线程完成
    for thread in all_threads:
        thread.join()
    
    print("所有生产和消费任务已完成")


if __name__ == "__main__":
    main()