进程间通信与线程

进程间通信与线程

一、进程间通过队列通信

"""
队列:先进先出
堆栈:先进后出
"""
from multiprocessing import Queue


q = Queue(5)  # 括号内可以传参数 表示的是这个队列的最大存储数
# 往队列中添加数据
q.put(1)
q.put(2)
# print(q.full())  # 判断队列是否满了
q.put(3)
q.put(4)
q.put(5)
# print(q.full())
# q.put(6)  # 当队列满了之后 再放入数据 不会报错 会原地等待 直到队列中有数据被取走(阻塞态)

print(q.get())
print(q.get())
print(q.get())
print(q.empty())  # 判断队列中的数据是否取完
print(q.get())
print(q.get())
print(q.empty())
# print(q.get_nowait())  # 取值 没有值不等待直接报错
# print(q.get())  # 当队列中的数据被取完之后 再次获取 程序会阻塞 直到有人往队列中放入值
"""
full
get_nowait
empty
都不适用于多进程的情况
"""

q = Queue(5)限定这个队列中最多存放5个值，不写则为默认值（这个值很大）
q.full()用于查看这个队列是否已满
q.empty()用于查看这个队列是否为空
q.put()用于往队列里面放值，如果队列已满，则原地等待队列中的值被取走（阻塞态）
q.get()用于从队列里面取值，如果队列已空，则原地等待有人向队列中放值（阻塞态）
q.get_nowait()与q.get()用途一致，区别在于，当队列里面没有值的时候，会直接报错

二、进程间通信IPC（inter process communication）

from multiprocessing import Process,Queue

def producer(q):
    q.put('hello GF~')

def consumer(q):
    print(q.get())

if __name__ == '__main__':
    q = Queue()
    p = Process(target=producer,args=(q,))
    c = Process(target=consumer, args=(q,))
    p.start()
    c.start()


"""
子进程放数据 主进程获取数据
两个子进程相互放 取数据
"""

三、生产者消费者模型

"""
生产者:生产/制造数据的
消费者:消费/处理数据的
例子:做包子的,买包子的
        1.做包子远比买包子的多
        2.做包子的远比包子的少
        供需不平衡的问题
"""
from multiprocessing import Process,Queue,JoinableQueue
import random
import time


def producer(name,food,q):
    for i in range(10):
        data = '%s生产了%s%s'%(name,food,i)
        time.sleep(random.random())
        q.put(data)
        print(data)

def consumer(name,q):
    while True:
        data = q.get()
        if data == None:break
        print('%s吃了%s'%(name,data))
        time.sleep(random.random())
        q.task_done()  # 告诉队列你已经从队列中取出了一个数据 并且处理完毕了



if __name__ == '__main__':
    q = JoinableQueue()
    
    p = Process(target=producer,args=('大厨egon','馒头',q))
    p1 = Process(target=producer,args=('跟班tank','生蚝',q))
    c = Process(target=consumer,args=('许兆龙',q))
    c1 = Process(target=consumer,args=('吃货jerry',q))
    p.start()
    p1.start()
    c.daemon = True
    c1.daemon = True
    c.start()
    c1.start()
    p.join()
    p1.join()

    q.join()  # 等待队列中数据全部取出

四、线程

1、什么是线程

• 进程线程其实都是虚拟单位，都是用来帮助我们形象的描述某种事物

• 进程：资源单位

• 线程：执行单位

  将内存比喻成工厂

  那么进程就相当于是工厂里面的车间

  而线程就相当于是车间里面的流水线

PS：每个进程都自带一个线程，线程才是真正的执行单位，进程只是在线程运行过程中提供代码运行所需要的资源

2、为什么要有线程

• 开进程

  申请内存空间，消耗资源

  拷贝代码，消耗资源

• 开线程

  一个进程内可以开多个线程，并且线程与线程之间数据时共享的

PS：开启线程的开销要远远小于开启进程的开销

五、创建线程的两种方式

# 方式一
from threading import Thread
import time


def task(name):
    print(f'{name} is running')
    time.sleep(1)
    print(f'{name} is over')


t = Thread(target=task, args=('egon',))
t.start()
print('主')
'''
开线程不需要在__main__代码块中执行，但是习惯性还是写在__main__代码块内
由于开启线程的开销要远远低于进程，所以会出现egon is running这句话在“主”之前打印
同样的，主线程的代码执行完毕后，也要等待子线程的代码执行完毕才能结束运行
因为主线程的结束意味着主进程的结束，而主进程一旦结束，主进程所占用的内存空间就会被操作系统回收，而子线程是在主进程中创建的，所以主线程在子线程之前结束，会导致子线程被强行终止。
'''

# 方式二
from threading import Thread
import time


class MyThread(Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(f'{self.name} is running')
        time.sleep(1)
        print(f'{self.name} is over')


p = MyThread('egon')
p.start()
print('主')
'''
与进程的创建类似，创建一个类继承Thread类，覆盖__init__方法，run()方法。
'''

六、线程对象及其他方法

from threading import Thread, current_thread, active_count
import time
import os


def task(name, i):
    print(f'{name} is running')
    print('子线程名', current_thread().name)
    print('子线程所处进程号', os.getpid())
    time.sleep(i)
    print(f'{name} is over')


if __name__ == '__main__':
    t = Thread(target=task, args=('egon', 2))
    t1 = Thread(target=task, args=('jason', 1))
    t.start()
    t1.start()
    t1.join()
    print('当前正在活跃的线程数', active_count())
    print('主')
    print('主线程号', current_thread().name)
    print('主线程所处进程号', os.getpid())
'''
current_thread()表示当前线程
current_thread().name查看当前线程名
active_count()统计当前正在活跃的线程数
'''

七、守护线程

from threading import Thread, current_thread
import time


def task(i):
    print(current_thread().name)
    time.sleep(i)
    print('GG')


if __name__ == '__main__':
    # for i in range(3):
    #     t = Thread(target=task, args=(i,))
    #     t.start()
    t = Thread(target=task, args=('egon',))
    t.daemon = True  # 将该线程设置为主线程的守护线程，一旦主线程运行结束，子线程也跟着结束
    t.start()
    print('主')
# Thread-1
# 主
'''
***GG不会打印***
主线程的结束也就意味着进程的结束
主线程必须等待其他非守护线程的结束才能结束
(意味子线程在运行的时候需要使用进程中的资源,而主线程一旦结束了资源也就销毁了)
'''

八、线程间通信

from threading import Thread

money = 100


def task():
    global money
    money = 666


t = Thread(target=task)
t.start()
t.join()
print(money)
'''
子线程修改了主线程的值，线程间的数据是共享的
'''

九、互斥锁

from threading import Thread, Lock
import time

n = 100


def task(mutex):
    global n
    mutex.acquire()
    temp = n
    time.sleep(0.1)  # 模拟网络延迟
    n = temp - 1
    mutex.release()


t_list = []
mutex = Lock()
for i in range(100):
    t = Thread(target=task, args=(mutex,))
    t.start()
    t_list.append(t)
for t in t_list:
    t.join()
print(n)
'''
多个线程同时操作同一个数据，则会出现数据错乱的现象
想到加锁，与进程中的互斥锁类似
'''

十、一道小题目

from threading import Thread
import time


def foo():
    print(123)
    time.sleep(1)
    print('end123')


def bar():
    print(456)
    time.sleep(2)
    print('end456')


if __name__ == '__main__':
    t1 = Thread(target=foo)
    t2 = Thread(target=bar)
    t1.daemon = True
    t1.start()
    t2.start()
    print('main')
'''
打印结果：
123
456
main
end123
end456

虽然将t1设为为守护线程了，但主线程代码运行结束以后并不会立即结束，而是会等待所有非守护线程结束之后才结束，在本例中，t2即为非守护线程，在它结束之前t1线程就已经结束了。如果t2线程结束之后，t1线程还未结束，那么end123这句话就不会打印。
'''