python之多线程多进程并发通信

╰+攻爆jí腚メ 2022-05-09 14:40 313阅读 0赞

## 1.独立的进程内存空间与共享的服务器进程空间 ##

进程之间是：互不干扰的独立内存空间

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    # @Time    : 2018/10/16 7:59
    # @Author  : DoubleChina
    # @Site    : 
    # @File    : ConcurrentTest.py
    # @Software: PyCharm
    # 并发通讯
    from multiprocessing import Process
    a = 1
    def func():
        global a
        a = 2
    p = Process(target=func())
    p.start()
    p.join()
    #输出的是1，并不是2，两个进程之间没有共享内存
    print(a)

解决了内存共享的问题

![在这里插入图片描述][70]

from multiprocessing import Process, Manager
    #启动服务器进程
    mgr = Manager()
    #list、dict、queue
    d = mgr.dict()
    # d = dict()
    print(type(d))
    
    
    def func(d):
        d['a'] = 'a'
    
    
    p = Process(target=func, args=(d,))
    p.start()
    p.join()
    #{'a': 'a'}  ,服务器进程间共享了数据
    print(d)

一般常用的空间类型是：

*  mgr.list()
 *  mgr.dict()
 *  mgr.Queue()

\#\#2.线程间共享的全局变量与同步锁的基本概念  
因为线程属于同一个进程，因此它们之间共享内存区域。因此全局变量是公共的。

import threading
    a = 1
    def func():
        global a
        a = 2
    t = threading.Thread(target=func)
    t.start()
    t.join()
    #输出2，共享了全局变量
    print(a)

#### 多线程同时访问一个变量，资源竞争导致数据异常 ####

# 2个线程分别执行1万次a+1
    from threading import Thread
    
    a = 0
    n = 100000
    
    
    def incr(n):
        global a
        for i in range(n):
            a = a + 1
    
    
    def dncr(n):
        global a
        for i in range(n):
            a = a - 1
    
    
    t1 = Thread(target=incr, args=(n,))
    t2 = Thread(target=dncr, args=(n,))
    t1.start()
    t2.start()
    
    t1.join()
    t2.join()
    #输出不一定是0
    print(a)

### 加锁解决资源竞争问题 ###

from threading import Thread, Lock
    a = 0
    n = 100000
    lock = Lock()  # 创建锁
    def incr(n):
        global a
        for i in range(n):
            lock.acquire()  # 获取锁
            a = a + 1
            lock.release()  # 释放锁
    def dncr(n):
        global a
        for i in range(n):
            lock.acquire()  # 获取锁，如果t1线程获取了锁，就会进入阻塞
            a = a - 1
            lock.release()  # 释放锁
    t1 = Thread(target=incr, args=(n,))
    t2 = Thread(target=dncr, args=(n,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    #输出是0
    print(a)

\#\#3.线程与进程安全的队列  
队列：一个入口，一个出口先入先出（FIFO）  
![在这里插入图片描述][70 1]

### 线程队列 ###

# 线程队列
    import queue
    
    q = queue.Queue(2)
    q.put(1)
    q.put(2)
    # q.put(3, block=False)  # 阻塞
    #获取队列
    print(q.get())
    # 测试队列是否满了
    print(q.full())
    # 判断队列是否是空的
    print(q.empty())
    # 队列大小
    print(q.qsize())
    # 等待队列完成
    print(q.join())
    # 任务结束：
    print(q.task_done())

### 进程队列 ###

# 进程队列
    from  multiprocessing import Queue, Process
    
    q = Queue()
    q.put(1)
    
    
    def func(q):
        # 队列有数据，共享资源
        print(q.get())
    
    
    p = Process(target=func, args=(q,))
    p.start()
    p.join()
    # print(q.get())

如果只是一个线程/进程在使用，那么它并不算公共资源。  
但是一旦多个线程/进程在同时使用，那么它就是一个公共资源。  
如果被当作公共资源使用，那么按理说是必须要加锁的。但是，线程安全或进程安全的队列中已经帮我们实现了锁。因此我们不需要再自己使用锁来同步。

## 4.消费者与生产者模式 ##

生产者与消费者模型：其实是把一个需要进程通信的问题分开考虑  
生产者，只需要往队列里面丢东西（生产者不需要关心消费者）  
消费者，只需要从队列里面拿东西（消费者也不需要关心生产者）  
![在这里插入图片描述][70 2]

import threading
    import random
    import queue
    import time
    
    
    class Producer(threading.Thread):
        def __init__(self, queue):
            super().__init__()
            self.queue = queue
    
        def run(self):
            for i in range(10):
                r = random.randint(0, 9)
                if not self.queue.full():
                    self.queue.put(r)
                    print('队列里面添加了一个数据{}'.format(r))
                    time.sleep(2)
                else:
                    print('满了')
    
    
    class Consumer(threading.Thread):
        def __init__(self, queue):
            super().__init__()
            self.queue = queue
    
        def run(self):
            while True:
                # if not self.queue.empty():  # 判断是否为空
                data = self.queue.get()
                time.sleep(2)
                print('从队列里面获取数据{}'.format(data))
                self.queue.task_done()  # 通知队列
    
    
    if __name__ == '__main__':
        q = queue.Queue()
        p1 = Producer(q)
        c1 = Consumer(q)
        p1.start()
        c1.start()
        p1.join()
        c1.join()
        q.join()

### 进程&线程使用队列 ###

# 普通的队列，不能用在多进程通讯
    import queue
    #在进程中使用Manager() 、 Queue()
    from multiprocessing import Queue,Manager

[70]: /images/20220506/70f495b8285b4eb6bd8650335f6395b9.png
[70 1]: /images/20220506/2d5086daef664a6b9174bcd4e12a1776.png
[70 2]: /images/20220506/c2d3019d49b1434b87c2ae4360fc8ee0.png