day31 进程间通讯,线程

Love The Way You Lie 2023-08-17 17:48 117阅读 0赞

### IPC ###

## 进程间相互通讯    ##

进程间是相互独立的,资源无法共享,当一个进程想要把数据给另外一个进程,就需要考虑IPC,可以通过管道(队列)之间通讯

队列:管道+锁 先进先出

堆栈:先进后出

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Queue
    
    q = Queue(5)
    # 括号内的参数,表示的是这个队列的储存数
    q.put(1)
    # 添加数据
    q.put(2)
    q.put(3)
    q.put(4)
    print(q.full())
    # 判断队列是否满了
    q.put(5)
    # q.put(6)
    # 当队列满了之后,在向里面添加值,不会出现报错,会在原地等待,进入阻塞状态,等待队列中有值被取走
    
    # print(q.__dict__)
    
    
    print(q.get())
    # 获取数据
    print(q.get())
    print(q.get())
    print(q.empty())
    # 判断队列中的数据取完
    print(q.get())
    
    print(q.get())
    print(q.get_nowait())
    # 取值,没有值不等待直接报错
    print(q.get())
    # 当队列中的数据被取完之后,再次获取,程序会阻塞,进入阻塞状态,在原地等待,直到有值被添加
    
    
    """
    full
    empty
    get_nowait
    都不适用于多进程的情况下
    刚判断完,就有值被添加或取出,或者还没来得及添加
    """

#### (验证进程间通讯) ####

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

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process,Queue
    
    def producer(q):
        q.put('hello world')
    
    
    def consumer(q):
        print(q.get())
    
    
    if __name__ == '__main__':
        q = Queue()
        p1 = Process(target=producer,args=(q,))
        p2 = Process(target=consumer,args=(q,))
        p1.start()
        p2.start()

## 生产者消费者模型  ##

模型 就是解决某个问题套路

产生数据的一方称之为生产者

处理数据的一方称之为消费者

例如: 爬虫 生活中到处都是这种模型

饭店 厨师就是生产者 你吃饭的人就是消费者

生产者和消费,处理速度不平衡,一方快一方慢,导致一方需要等待另一方

原本,双方是耦合 在一起,消费必须等待生产者生成完毕在开始处理, 反过来

如果消费消费速度太慢,生产者必须等待其处理完毕才能开始生成下一个数据

将双方分开来.一专门负责生成,一方专门负责处理

这样一来数据就不能直接交互了 双方需要一个共同的容器

生产者完成后放入容器,消费者从容器中取出数据

这样就解决了双发能力不平衡的问题,做的快的一方可以继续做,不需要等待另一方

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from multiprocessing import Process,JoinableQueue
    import random
    import time
    
    def producer(name,food,q):
        for i in range(5):
            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()
        p1 = Process(target=producer,args=('小明','葱油饼',q))
        p2 = Process(target=producer,args=('小于','菜包',q))
        c1= Process(target=consumer,args=('胖丁',q))
        c2 = Process(target=consumer,args=('丫丫',q))
        p1.start()
        p2.start()
        c1.daemon = True
        c2.daemon = True
        c1.start()
        c2.start()
        p1.join()
        p2.join()
    
        q.join()  # 等到队列中数据全部取出
        # q.put(None)
        # q.put(None)

# 线程 #

线程是操作系统最小的运算调度单位,被包含在进程中,一个线程就是一个固定的 执行流程

### 线程的进程的关系  ###

线程不能单独存在 必须存在于进程中,

进程是一个资源单位,其包含了运行程序所需的所有资源

线程才是真正的执行单位

没有线程,进程中的资源无法被利用起来,所以一个进程至少包含一个线程,称之为主线程

当我们启动一个程序时,操作系统就会自己为这个程序创建一个主线程

线程可以由程序后期开启 ,自己开启线程称之为子线程

进程:资源单位
    线程:执行单位
    每个进程都自带一个线程,线程才是真正的执行单位,进程只是在线程运行过程中
        提供代码运行所需要的资源
    为什么要有线程
        开进程
            1.申请内存空间  耗资源
            2."拷贝代码"    耗资源
    
        开线程
            一个进程内可以起多个线程,并且线程与线程之间数据是共享的
        ps:开启线程的开销要远远小于开启进程的开销

## 创建线程 ##

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread
    import time
    使用方法一  直接实例化Thread类
    def task(name):
        print('%s is running'%name)
        time.sleep(1)
        print('%s is over'%name)
    # 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内
    t = Thread(target=task,args=('dada',))
    t.start()
    # 告诉操作系统开辟一个线程  线程的开销远远小于进程
    print('主')

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread
    import time
    使用方法二 继承Thread 覆盖run方法
    class MyThread(Thread):
        def __init__(self,name):
            super().__init__()
            self.name = name
    
        def run(self):
            print('%s is running'%self.name)
            time.sleep(3)
            print('%s is over'%self.name)
    
    t = MyThread('cc')
    t.start()
    print('主')

其他方法

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread,current_thread,active_count
    import time
    import os
    
    def task(name,i):
        print('%s is running'%name)
        # print('子current_thread:',current_thread().name)
        # print('子',os.getpid())
        time.sleep(i)
    
        print('%s is over'%name)
    # 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内
    t = Thread(target=task,args=('egon',1))
    t1 = Thread(target=task,args=('jason',2))
    t.start()  # 告诉操作系统开辟一个线程  线程的开销远远小于进程
    t1.start()  # 告诉操作系统开辟一个线程  线程的开销远远小于进程
    t1.join()  # 主线程等待子线程运行完毕
    print('当前正在活跃的线程数',active_count())
    # 小的代码执行完 线程就已经开启了
    print('主')
    # print('主current_thread:',current_thread().name)
    # print('主',os.getpid())

# 守护线程 #

一个线程可以设置为另一个线程的守护线程

特点: 被守护线程结束后守护线程也随之结束

守护线程会等到所有非守护线程结束后结束 ! 前提是除了主线程之外 还有后别的非守护  
当然如果守护线程已经完成任务 立马就结束了

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread,current_thread
    import time
    
    
    
    def task(i):
        print(current_thread().name)
        time.sleep(i)
        print('GG')
    # for i in range(3):
    #     t = Thread(target=task,args=(i,))
    #     t.start()
    t = Thread(target=task,args=(1,))
    t.daemon = True
    t.start()
    print('主')

### 验证线程间通讯 ###

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread
    
    
    money = 666
    
    def task():
        global money
        money = 999
    
    t = Thread(target=task)
    t.start()
    t.join()
    print(money)

### 互斥锁 ###

共享意味着竞争

线程中也存在安全问题,

多线程可以并发执行,一旦并发了并且访问了同一个资源就会有问题

解决方案:还是互斥锁

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread,Lock
    import time
    
    
    n = 100
    
    def task(mutex):
        global  n
        上锁
        mutex.acquire()
        tmp = n
        time.sleep(0.1)
        n = tmp - 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)

思考

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from threading import Thread
    from multiprocessing import Process
    import time
    def foo():
        print(123)
        time.sleep(1)
        print("end123")
    
    def bar():
        print(456)
        time.sleep(3)
        print("end456")
    
    if __name__ == '__main__':
        t1=Thread(target=foo)
        t2=Thread(target=bar)
        t1.daemon=True
        t1.start()
        t2.start()
        print("main-------")

转载于:https://www.cnblogs.com/komorebi/p/11342926.html

[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20230808/7dee8a2b49f145f7a022f3ee89f0d933.png