Python并发编程：实战案例探讨

原创柔光的暖阳◎ 2025-01-09 19:27 10阅读 0赞

Python的并发编程主要是通过多线程（threading）和异步IO（asyncio）来实现。下面我们将通过几个实战案例，深入理解和实践Python的并发编程。

1. **多线程**：
   - **生产者消费者模型**：创建一个生产队列（Producer）和消费队列（Consumer），线程交替进行生产任务和消费任务。
   
   ```python
   import threading

class Queue:
       def __init__(self, max_size=10):
           self.max_size = max_size
           self.queue = []

def put(self, item):
           if len(self.queue) < self.max_size:
               self.queue.append(item)
           else:
               print("Queue is full!")

def get(self):
           if not self.queue:
               return None
           return self.queue.pop(0)

producer_queue = Queue()
   consumer_queue = Queue()

threading.Thread(target=producer, args=(producer_queue,), daemon=True).start()
   threading.Thread(target=consumer, args=(consumer_queue,), daemon=True).start()

# 生产者任务
   def producer(queue):
       for _ in range(10):
           item = "Item %d" % (_ + 1))
           queue.put(item)

# 消费者任务
   def consumer(queue):
       while True:
           try:
               item = queue.get()
               print(f"Consumed: {item}")
           except queue.Empty:
               break

```
   
2. **异步IO**：
   - **asyncio多线程示例**：使用asyncio和threading，实现一个并发的HTTP请求。
   
   ```python
   import asyncio
   import aiohttp

async def fetch(session, url):
       async with session.get(url) as response:
           return await response.text()

urls = ["https://www.google.com", "https://www.yahoo.com"]
   loop = asyncio.get_event_loop()
   tasks = [fetch(aiohttp.Client(), url)) for url in urls]
   
   results = loop.run_until_complete(asyncio.gather(*tasks)))
   print(results)

```
   
以上就是Python并发编程的一些实战案例，希望对你理解和实践Python并发编程有所帮助。