Python队列、进程间通信、线程案例

# 并发运行，效率高，但竞争写同一文件，数据写入错乱
# 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()

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')

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('主')

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)