进程互斥锁

多进程同时抢购余票

# 并发运行，效率高，但竞争写同一文件，数据写入错乱# data.json文件内容为 {"ticket_num": 1}import jsonimport timefrom multiprocessing import Processdef 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 jsonimport timefrom multiprocessing import Process, Lockdef 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()

进程互斥锁：

让并发变成串行，牺牲了执行效率，保证了数据安全

在程序并发时，需要修改数据使用

队列

队列遵循的是先进先出

队列：相当于内存中一个队列空间，可以存放多个数据，但数据的顺序是由先进去的排在前面。

q.put() 添加数据

q.get() 取数据，遵循队列先进先出

q.get_nowait() 获取队列数据， 队列中没有就会报错

q.put_nowait 添加数据，若队列满了也会报错

q.full() 查看队列是否满了

q.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)

进程间数据是相互隔离的，若想实现进程间通信，可以利用队列

from multiprocessing import Process, Queuedef 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, Queueimport 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('主')

线程

线程的概念

进程与线程都是虚拟单位

进程：资源单位

线程：执行单位

开启一个进程，一定会有一个线程，线程才是真正执行者

开启进程：

• 开辟一个名称空间，每开启一个进程都会占用一份内存资源
• 会自带一个线程

开启线程：

• 一个进程可以开启多个线程
• 线程的开销远小于进程

注意：线程不能实现并行，线程只能实现并发，进程可以实现并行

线程的两种创建方式

from threading import Threadimport time# 创建线程方式1def task(): print('线程开启') time.sleep(1) print('线程结束')if __name__ == '__main__': t = Thread(target=task) t.start()# 创建线程方式2class MyThread(Thread): def run(self): print('线程开启...') time.sleep(1) print('线程结束...')if __name__ == '__main__': t = MyThread() t.start()

线程对象的方法

from threading import Threadfrom threading import current_threadimport timedef 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 Threadfrom threading import Lockimport timemutex = Lock()n = 100def 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)

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。