Python并发编程实战：进程锁、队列通信与线程详解
本文深入解析Python并发编程核心机制。详细讲解如何利用进程互斥锁解决多进程抢票数据竞争问题，确保数据一致性；介绍Queue队列在进程间通信（IPC）中的应用，包括生产者-消费者模型实现；对比进程与线程的资源开销差异，演示两种线程创建方式、常用方法以及使用互斥锁保护共享数据的具体代码示例。
进程互斥锁与数据同步
进程互斥锁
多进程抢票场景模拟
在多进程并发环境下模拟多人同时抢购剩余票数的场景。当多个进程同时读取和修改同一资源文件时，若缺乏同步机制，极易导致数据写入错乱或超卖现象。
# 并发运行，效率高，但竞争写同一文件，数据写入错乱
# data.json文件内容为 {"ticket_num": 1}
import json
import time
from multiprocessing import Process
def search(user):
with open('data.json', 'r', encoding='utf-8') as f:
dic = json.load(f)
print(f'用户{user}查看余票，还剩{dic.get("ticket_num")}...')
def buy(user):
with open('data.json', 'r', encoding='utf-8') as f:
dic = json.load(f)
time.sleep(0.1)
if dic['ticket_num'] > 0:
dic['ticket_num'] -= 1
with open('data.json', 'w', encoding='utf-8') as f:
json.dump(dic, f)
print(f'用户{user}抢票成功!')
else:
print(f'用户{user}抢票失败')
def run(user):
search(user)
buy(user)
if __name__ == '__main__':
for i in range(10): # 模拟10个用户抢票
p = Process(target=run, args=(f'用户{i}', ))
p.start()锁机制保障数据安全
引入锁机制是保障并发环境下数据安全的必要手段。通过互斥锁，可以确保同一时刻只有一个进程能修改共享资源，从而避免数据竞争。
# data.json文件内容为 {"ticket_num": 1}
import json
import time
from multiprocessing import Process, Lock
def search(user):
with open('data.json', 'r', encoding='utf-8') as f:
dic = json.load(f)
print(f'用户{user}查看余票，还剩{dic.get("ticket_num")}...')
def buy(user):
with open('data.json', 'r', encoding='utf-8') as f:
dic = json.load(f)
time.sleep(0.2)
if dic['ticket_num'] > 0:
dic['ticket_num'] -= 1
with open('data.json', 'w', encoding='utf-8') as f:
json.dump(dic, f)
print(f'用户{user}抢票成功!')
else:
print(f'用户{user}抢票失败')
def run(user, mutex):
search(user)
mutex.acquire() # 加锁
buy(user)
mutex.release() # 释放锁
if __name__ == '__main__':
# 调用Lock()类得到一个锁对象
mutex = Lock()
for i in range(10): # 模拟10个用户抢票
p = Process(target=run, args=(f'用户{i}', mutex))
p.start()进程互斥锁的核心作用在于将并发操作转化为串行执行。虽然这在一定程度上牺牲了程序的执行效率，但它严格保证了数据的安全性和一致性。
互斥锁通过强制串行化关键代码段，以性能换取数据正确性。在涉及共享资源修改的高并发场景中，这是防止数据损坏的关键策略。
在程序进行并发操作且涉及数据修改时，必须使用锁机制来保护共享资源，防止多个进程同时写入导致的状态不一致。
Queue队列基础与操作
队列（Queue）是Python中用于进程间通信的重要数据结构。
队列遵循先进先出（FIFO）原则，即最早进入队列的数据将最先被取出。
队列可视为内存中一个具有容量限制的空间，用于存储多个数据对象。数据的存取顺序严格由进入队列的先后顺序决定，确保处理逻辑的可预测性。
使用 q.put() 方法向队列中添加数据。如果队列已满，该方法默认会阻塞等待，直到有空闲位置。
使用 q.get() 方法从队列中获取数据，该操作严格遵循先进先出原则。如果队列为空，该方法会阻塞等待，直到有数据可用。
使用 q.get_nowait() 尝试获取队列数据。如果队列中没有数据，该方法不会阻塞，而是立即抛出 Empty 异常。
使用 q.put_nowait() 尝试添加数据。如果队列已满，该方法不会阻塞，而是立即抛出 Full 异常。
使用 q.full() 方法检查队列是否已满，返回布尔值。
使用 q.empty() 方法检查队列是否为空，返回布尔值。注意：由于并发环境下的时序问题，empty() 的返回值可能不准确，仅作为参考。
from multiprocessing import Queue
# 调用队列类，实例化队列对象
q = Queue(5) # 队列中存放5个数据
# put添加数据，若队列里的数据满了就会卡住
q.put(1)
print('进入数据1')
q.put(2)
print('进入数据2')
q.put(3)
print('进入数据3')
q.put(4)
print('进入数据4')
q.put(5)
print('进入数据5')
# 查看队列是否满了
print(q.full())
# 添加数据, 若队列满了也会报错
q.put_nowait(6)
# q.get() 获取的数据遵循先进先出
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
# print(q.get())
print(q.get_nowait()) # 获取队列数据， 队列中没有就会报错
# 判断队列是否为空
print(q.empty())
q.put(6)
print('进入数据6')进程间通信（IPC）与队列应用
进程间通信（Inter-Process Communication, IPC）是指不同进程之间交换数据或信号的技术。
IPC 是操作系统提供的机制，允许独立运行的进程之间进行数据交换。
由于进程间内存空间相互隔离，无法直接访问对方变量。要实现进程间通信，可以利用 Queue 队列作为中介，安全地传递数据。
from multiprocessing import Process, Queue
def task1(q):
data = 'hello 你好'
q.put(data)
print('进程1添加数据到队列')
def task2(q):
print(q.get())
print('进程2从队列中获取数据')
if __name__ == '__main__':
q = Queue()
p1 = Process(target=task1, args=(q, ))
p2 = Process(target=task2, args=(q, ))
p1.start()
p2.start()
print('主进程')生产者-消费者模型
生产者与消费者模型是并发编程中经典的队列应用场景，用于解耦数据生成与处理过程。
在程序设计中，通过队列实现生产者与消费者模式：生产者进程负责将数据添加到队列中，消费者进程则从队列中获取并处理数据，两者通过队列进行解耦协作。
from multiprocessing import Process, Queue
import time
# 生产者
def producer(name, food, q):
for i in range(10):
data = food, i
msg = f'用户{name}开始制作{data}'
print(msg)
q.put(data)
time.sleep(0.1)
# 消费者
def consumer(name, q):
while True:
data = q.get()
if not data:
break
print(f'用户{name}开始吃{data}')
if __name__ == '__main__':
q = Queue()
p1 = Process(target=producer, args=('neo', '煎饼', q))
p2 = Process(target=producer, args=('wick', '肉包', q))
c1 = Process(target=consumer, args=('cwz', q))
c2 = Process(target=consumer, args=('woods', q))
p1.start()
p2.start()
c1.daemon = True
c2.daemon = True
c1.start()
c2.start()
print('主')Python线程基础与创建
线程（Thread）是操作系统能够进行运算调度的最小单位，常被包含在进程之中。
理解线程的基本概念及其与进程的异同，是掌握Python并发编程的基础。
进程和线程都是操作系统引入的虚拟概念，用于更好地管理资源分配和执行调度。
进程是资源分配的基本单位，每个进程拥有独立的内存空间、文件描述符等资源。
线程是CPU调度和执行的基本单位，线程本身几乎不拥有系统资源，仅需要少量的栈空间和寄存器。
每当创建一个进程时，操作系统会自动为其创建一个主线程。因此，一个进程至少包含一个线程，而真正的代码执行是由线程完成的。
进程与线程的创建开销对比
创建进程的过程包括：
- 开辟一个新的名称空间，为进程分配独立的内存地址空间。每开启一个进程都会占用一份完整的内存资源，开销较大。
- 系统会自动为新建的进程创建一个主线程，该线程负责执行进程的主要逻辑。
创建线程的过程包括：
- 一个进程可以创建多个线程，这些线程共享进程的资源（如内存空间、文件句柄等）。
- 线程的创建和切换开销远小于进程，因为无需分配新的内存空间和资源副本，因此更适合轻量级并发任务。
线程并发与进程并行的区别
需要注意：由于GIL（全局解释器锁）的存在，Python中的线程无法实现真正的并行计算（多核CPU同时执行），只能实现并发（交替执行）。而进程可以实现真正的并行，利用多核CPU优势。
线程的两种创建方式
Python中创建线程主要有两种常见方式：
from threading import Thread
import time
# 创建线程方式1
def task():
print('线程开启')
time.sleep(1)
print('线程结束')
if __name__ == '__main__':
t = Thread(target=task)
t.start()
# 创建线程方式2
class MyThread(Thread):
def run(self):
print('线程开启...')
time.sleep(1)
print('线程结束...')
if __name__ == '__main__':
t = MyThread()
t.start()线程对象常用方法
线程对象提供了一系列方法用于控制和管理线程状态，如启动、等待、判断存活等。
from threading import Thread
from threading import current_thread
import time
def task():
print(f'线程开启{current_thread().name}')
time.sleep(1)
print(f'线程结束{current_thread().name}')
if __name__ == '__main__':
t = Thread(target=task)
print(t.isAlive())
# t.daemon = True
t.start()
print(t.isAlive())线程互斥锁与共享数据保护
线程互斥锁用于解决多线程环境下对共享资源访问的竞争问题，确保数据一致性。
与进程不同，同一进程内的所有线程共享该进程的内存空间和数据，因此线程间数据是共享的，这也带来了数据竞争的风险。
from threading import Thread
from threading import Lock
import time
mutex = Lock()
n = 100
def task(i):
print(f'线程{i}启动')
global n
mutex.acquire()
temp = n
time.sleep(0.1)
n = temp - 1
print(n)
mutex.release()
if __name__ == '__main__':
t_l = []
for i in range(100):
t = Thread(target=task, args=(i, ))
t_l.append(t)
t.start()
for t in t_l:
t.join()
print(n)以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
