python_并发与通信

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

知识点一：进程间通信的限制

进程是独立的，互不干扰的独立内存空间
我们想不能修改变量
但是，深层次问题是，这个进程与那个进程完全失去了联系

import multiprocessing

a = 1 # 全局变量

def func():

    global a   # 声明全局

    a += 1   # 修改全局变量

p = multiprocessing.Process(target=func)

p.start()

p.join()  # 等待子进程结束

print(a)    # 结果是1  不同的进程

执行结果：

1

知识点二：进程间通信的解决方案

1. 管理器负责与公共进程通信
2. 代理负责操作共享的空间

知识点三： Manger对象的基本使用

一般常用的空间类型是：
1. mgr.list()
2. mgr.dict()
3. mgr.Queue()

关于Queue()我们稍后演示

import multiprocessing

from multiprocessing import Manager

'''

常用的空间类型：

1、mgr.list() 列表

2、mgr.dict() 字典

3、mgr.Queue() 队列

'''

mgr = Manager()  # 创建一个服务器进程, 并返回与其通信的管理系

list_proxy = mgr.list()   # 在服务器进程中开辟一个列表空间，并在主进程中返回一个代理

print(type(list_proxy))  # ListProxy

def func(listData):

    test = 'abcdef'

    for i in test:

        listData.append(i)

p = multiprocessing.Process(target=func, args=(list_proxy,))

p.start()

p.join()  # 等待进程

print(list_proxy)

线程间共享的全局变量与同步锁的基本概念

线程间全局变量的共享
提示！
因为线程属于同一个进程，因此它们之间共享内存区域。

因此全局变量是公共的。

import threading

a = 1

def func():

    global a

    a = 2

t = threading.Thread(target=func)

t.start()

t.join()

print(a)  # 结果是2   一个进程下的多个线程

共享内存间存在竞争问题

提示！
如果1000000不能出现效果
可以继续在后面加0

import threading

a = 1

n = 10000

def func(n):

    global a

    for i in range(n):

        # print("加法:",i)

        a += 1

def func2(n):

    global a

    for i in range(n):

        # print("减法:",i)

        a -= 1

t1 = threading.Thread(target=func, args=(1000000,))   # 如果是1000000   结果就不是1了， 因为共享内存空间存在竞争问题

t2 = threading.Thread(target=func2, args=(1000000,))    # 抢占资源，轮询调用的

t1.start()

t2.start()

t1.join()

t2.join()

print(a)  # 结果是2   一个进程下的多个线程

使用锁来控制共享资源的访问
方法一：

import threading

from threading import Lock

'''

线程锁: lock()

'''

lock = Lock()  # 生成一个锁的实例

a = 1

n = 10000

def func(n):

    global a

    for i in range(n):

        with lock:

            a += 1

def func2(n):

    global a

    for i in range(n):

        with lock:

            a -= 1

t1 = threading.Thread(target=func, args=(1000000,))

t2 = threading.Thread(target=func2, args=(1000000,))

t1.start()

t2.start()

t1.join()

t2.join()

print(a)  # 结果是1

方法二：

import threading

from threading import Lock

'''

线程锁: lock()

'''

lock = Lock()  # 生成一个锁的实例

a = 1

n = 10000

def func(n):

    global a

    for i in range(n):

        lock.acquire()  # 上锁

        a += 1

        lock.release()  # 解锁

def func2(n):

    global a

    for i in range(n):

        lock.acquire()  # 上锁

        a -= 1

        lock.release()  # 解锁

t1 = threading.Thread(target=func, args=(1000000,))

t2 = threading.Thread(target=func2, args=(1000000,))

t1.start()

t2.start()

t1.join()

t2.join()

print(a)  # 结果是1

线程与进程安全的队列

队列的基本概念
一个入口，一个出口
先入先出（FIFO）

queue.Queue
线程安全队列
操作一览

from queue import Queue

# 线程队列

q = Queue(5)  # 队列里有5个长度，最大容纳，超出会溢出

q.put('a')  # 入队

print("队列长度为: ",q.qsize()) # 队列长度

print("测试空: ", q.empty())  # 测试空  近似

print("测试结束: ", q.task_done())  # 任务结束

print("测试满: ", q.full())  # 测试满  近似

print("等待完成: ", q.join())  # 等待完成

print(q.get())   # 出队

结果：

队列长度为:  1

测试空:  False

测试结束:  None

测试满:  False

等待完成:  None

a

mgr.Queu
线程安全队列
操作一览

from multiprocessing import Manager

mgr = Manager()

qq = mgr.Queue()

qq.put('a')  # 入队

print("测试满: ", qq.full())  # 测试满  近似

print("队列长度为: ",qq.qsize()) # 队列长度  近似

print("测试空: ", qq.empty())  # 测试空  近似

print("出队: ", qq.get())

结果：

测试满:  False

队列长度为:  1

测试空:  False

出队:  a

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

消费者与生产者模式

问题一：什么是消费者与生产者模式？

所谓，生产者与消费者模型，其实是把一个需要进程通信的问题分开考虑
生产者，只需要往队列里面丢东西（生产者不需要关心消费者）
消费者，只需要从队列里面拿东西（消费者也不需要关心生产者）

问题二：为什么需要消费者与生产者模式？

消费者与生产者模式的应用
—
Web服务器与Web框架之间的关系

问题三：如何通过队列实现消费者与生产者模式？

多线程的消费者与生产者模式

方法一：

from threading import Thread

from queue import Queue

import random

q = Queue(5)

# 生成者

'''

只关心队列是否已满，没满则生成，满了就阻塞

'''

class Produce(Thread):

    def __init__(self, queue):

        super().__init__()  # 重写 init

        self.queue = queue

    def run(self):

        while True:

            item = random.randint(0, 99)

            self.queue.put(item)     # 只要队列没满，向队列中存入数据

            print("生产者-->已生产 %s， 并将其加入到了队列中" % item)

# 消费者

'''

只关心队列是否为空。 不为空，则消费，为空则阻塞

'''

class Consumer(Thread):

    def __init__(self, queue):

        super().__init__()

        self.queue = queue

    def run(self):

        while True:

            item = self.queue.get()   # 只要队列不为空，就从队列取出数据

            print("消费者：从队列中取出 %s " % item)

            self.queue.task_done()

p = Produce(q)

c = Consumer(q)

p.start()

c.start()

结果：

生产者-->已生产 28， 并将其加入到了队列中

生产者-->已生产 2， 并将其加入到了队列中

生产者-->已生产 64， 并将其加入到了队列中

生产者-->已生产 57， 并将其加入到了队列中

生产者-->已生产 94， 并将其加入到了队列中

生产者-->已生产 52， 并将其加入到了队列中

消费者：从队列中取出 28

生产者-->已生产 53， 并将其加入到了队列中

消费者：从队列中取出 2

生产者-->已生产 72， 并将其加入到了队列中

消费者：从队列中取出 64

消费者：从队列中取出 57

生产者-->已生产 29， 并将其加入到了队列中

消费者：从队列中取出 94

生产者-->已生产 46， 并将其加入到了队列中

生产者-->已生产 70， 并将其加入到了队列中

消费者：从队列中取出 52

生产者-->已生产 11， 并将其加入到了队列中

消费者：从队列中取出 53

生产者-->已生产 55， 并将其加入到了队列中

消费者：从队列中取出 72

生产者-->已生产 83， 并将其加入到了队列中

消费者：从队列中取出 29

生产者-->已生产 0， 并将其加入到了队列中

消费者：从队列中取出 46

生产者-->已生产 66， 并将其加入到了队列中

消费者：从队列中取出 70

生产者-->已生产 86， 并将其加入到了队列中

消费者：从队列中取出 11

生产者-->已生产 30， 并将其加入到了队列中

消费者：从队列中取出 55

生产者-->已生产 48， 并将其加入到了队列中

消费者：从队列中取出 83

生产者-->已生产 41， 并将其加入到了队列中

消费者：从队列中取出 0

生产者-->已生产 67， 并将其加入到了队列中

方法二:

from multiprocessing import Process, Manager

import random

m = Manager()

q = m.Queue(5)

# 生成者

'''

只关心队列是否已满，没满则生成，满了就阻塞

'''

class Produce(Process):

    def __init__(self, queue):

        super().__init__()  # 重写 init

        self.queue = queue

    def run(self):

        while True:

            item = random.randint(0, 99)

            self.queue.put(item)     # 只要队列没满，向队列中存入数据

            print("生产者-->已生产 %s， 并将其加入到了队列中" % item)

# 消费者

'''

只关心队列是否为空。 不为空，则消费，为空则阻塞

'''

class Consumer(Process):

    def __init__(self, queue):

        super().__init__()

        self.queue = queue

    def run(self):

        while True:

            item = self.queue.get()   # 只要队列不为空，就从队列取出数据

            print("消费者：从队列中取出 %s " % item)

            self.queue.task_done()

p = Produce(q)

c = Consumer(q)

p.start()

c.start()

p.join()

c.join()

结果：

生产者-->已生产 96， 并将其加入到了队列中

生产者-->已生产 53， 并将其加入到了队列中

生产者-->已生产 62， 并将其加入到了队列中

生产者-->已生产 21， 并将其加入到了队列中

生产者-->已生产 5， 并将其加入到了队列中

生产者-->已生产 83， 并将其加入到了队列中

消费者：从队列中取出 96

生产者-->已生产 54， 并将其加入到了队列中

消费者：从队列中取出 53

生产者-->已生产 99， 并将其加入到了队列中

消费者：从队列中取出 62

生产者-->已生产 93， 并将其加入到了队列中

消费者：从队列中取出 21

生产者-->已生产 58， 并将其加入到了队列中

总结完毕.

作者：含笑半步颠√

博客链接：https://www.cnblogs.com/lixy-88428977

声明：本文为博主学习感悟总结，水平有限，如果不当，欢迎指正。如果您认为还不错，欢迎转载。转载与引用请注明作者及出处。

巴特西

python_并发与通信

最新文章

热门文章