多进程之数据安全问题

目录

• 守护进程
• 进程同步的安全问题
• 进程通信
  • 今天主要说一下，第二种方案，
    • 第一种方法：Manager
      • 应用：
    • 内存的第二种方法：队列（推荐使用）
      • 创建队列的类（底层就是以管道和锁定的方式实现）
      • 参数介绍：
      • 方法介绍（主要方法）：
      • 其他方法（了解）：
      • 应用：
    • 内存的第3种方法：管道（不推荐使用，了解即可）
      • 方法：
      • 应用：

守护进程

主进程创建守护进程

　　其一：守护进程会在主进程代码执行结束后就终止

　　其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

import multiprocessing
import time

def taks_one():
    print("zi run")
    time.sleep(3)
    print("zi over")

if __name__ == '__main__':
    print("父 start")
    p = multiprocessing.Process(target=taks_one)
    p.daemon = True		# 在进程开启前，设置守护进程
    p.start()
    time.sleep(5)
    print("父 over")

以上代码中，主进程相当于被守护进程，子进程相当于守护进程

进程同步的安全问题

问题：

import multiprocessing
import time


def task1():
    for i in range(1000):
        print("------------name is nick")
        print("------------gender is girl")
        print("------------age is 18")


def task2():
    for i in range(1000):
        print("++++++++++++++name is bgon")
        print("++++++++++++++gender is oldman")
        print("++++++++++++++age is 48")

if __name__ == '__main__':
    pass
    p1 = multiprocessing.Process(target=task1)
    p2 = multiprocessing.Process(target=task2)

    p1.start()
    p2.start()

执行以上代码时，可能会存在task1函数中的打印和task2函数中的打印交叉的情况。如果将task1中的打印看做是一次完整的打印，这样就会造成task1中只打印了一半，然后就会打印task2中的信息。对于我们而言，这并不是我们想要看到的结果。

解决办法一：加join,使其变为串行

import multiprocessing
import time


def task1():
    for i in range(1000):
        print("------------name is nick")
        print("------------gender is girl")
        print("------------age is 18")


def task2():
    for i in range(1000):
        print("++++++++++++++name is bgon")
        print("++++++++++++++gender is oldman")
        print("++++++++++++++age is 48")

if __name__ == '__main__':
    pass
    p1 = multiprocessing.Process(target=task1)
    p2 = multiprocessing.Process(target=task2)

    p1.start()
    p1.join()
    p2.start()

优点：

1. 解决了进程安全问题

缺点：

1. 效率变低

解决办法二：增加互斥锁

import multiprocessing
import time,random

# 创建锁
mutext = multiprocessing.Lock()

def task1(mutext):
    # 并发
    for i in range(1000):
        print(1)

    # 串行
    mutext.acquire()    # 加锁
    print("------------name is nick")
    time.sleep(random.random())
    print("------------gender is girl")
    print("------------age is 18")
    mutext.release()    # 解锁

def task2(mutext):
    # 并发
    for i in range(1000):
        print(2)

    # 串行
    mutext.acquire()    # 加锁
    print("++++++++++++++name is bgon")
    time.sleep(random.random())
    print("++++++++++++++gender is oldman")
    print("+++++++++++++age is 48")
    mutext.release()    # 解锁

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=task1,args=(mutext,))	# 传参（重点）
    p2 = multiprocessing.Process(target=task2,args=(mutext,))	# 传参（重点）

    p1.start()
    p2.start()

锁其实只是给执行代码加了限制，本质是一个标志为True或False

lock = False

def task1():
    global lock
    if lock = False:
        lock = True
        # 代码块1
        lock = False
        
def task2():
    global lock
    if lock = False:
        lock = True
        # 代码块2
        lock = False

优点：

1. 可以让子进程中的部分代码变成串行，这样解决了进程安全问题
2. 没有加锁的部分，依然可以并发。用户选择权更多

PS：需要注意，如果你想要多个子进程中的部分代码串行，必须是同一把锁才能实现。

进程通信

a进程如果想要和b继承通信，应该怎么做？

1. 方案一：创建一个共享文件

   缺点：硬盘上数据，读写I/O不高，效率低

   优点：理论上交换的数据量可以非常大

   适用场景：交互不频繁，且数据量较大的情况。

2. 方案二：共享内存（重点）

   缺点：数据量不能太大

   优点：效率高

   适用场景：交互频繁，但是数据量小

3. 方案三：管道

   缺点：基于文件，它是单向的，比较复杂

4. 方案四：socket

   缺点：编程复杂，用在这个地方有些大材小用

   适用场景：基于网络交换数据

今天主要说一下，第二种方案，

第一种方法：Manager

应用：

import multiprocessing

def task1(data):
    data["age"] = 19

if __name__ == '__main__':
    data = {"age":20}
    m = multiprocessing.Manager()	# 创建Manager对象	
    m_data = m.dict(data)			# 创建公共字典，除了字典外，还可以共享很多其他的数据类型
    p = multiprocessing.Process(target=task1,args=(m_data,))
    p.start()
    p.join()	# 必须要加上join()
    print(m_data)
    print(data)

{'age': 19}
{'age': 20}

内存的第二种方法：队列（推荐使用）

进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的

创建队列的类（底层就是以管道和锁定的方式实现）

Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

参数介绍：

maxsize是队列中允许最大项数，省略则无大小限制。 

方法介绍（主要方法）：

q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。

q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
 
q.get_nowait():同q.get(False)

q.put_nowait():同q.put(False)

q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。

q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。

q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

其他方法（了解）：

q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞

q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。

q.join_thread()：连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

应用：

import multiprocessing


def task1(data):
    d = data.get()
    print(d)
    
if __name__ == '__main__':
    q = multiprocessing.Queue()		# 创建队列
    for i in range(1,11):
        q.put(i)				# 向队列中存放数据
    print(q.empty())			# 打印队列是否是空，False

    a_list = []		
    # 消费队列中的数据
    for i in range(10):
        p = multiprocessing.Process(target=task1,args=(q,))	# 将队列传参到子进程中
        a_list.append(p)
        p.start()

    for i in a_list:
        i.join()

    print(q.empty())			# 打印队列是否是空，True
    print(q)					

内存的第3种方法：管道（不推荐使用，了解即可）

方法：

#创建管道的类：
Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道

#参数介绍：
dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。
#主要方法：
    conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。
    conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
 #其他方法：
conn1.close():关闭连接。如果conn1被垃圾回收，将自动调用此方法
conn1.fileno():返回连接使用的整数文件描述符
conn1.poll([timeout]):如果连接上的数据可用，返回True。timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达。
 
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常。
conn.send_bytes(buffer [, offset [, size]])：通过连接发送字节数据缓冲区，buffer是支持缓冲区接口的任意对象，offset是缓冲区中的字节偏移量，而size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv_bytes()函数进行接收    
 
conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常。

应用：

import multiprocessing

def consumer(p,name):
    left,right=p
    left.close()		# 关闭左边的管道，右边只负责接受
    while True:
        try:
            baozi=right.recv()		# 右边接受数据
            print('%s 收到包子:%s' %(name,baozi))
        except EOFError:
            right.close()		# 异常则关闭连接
            break

def producer(seq,p):
    left,right=p
    right.close()		# 关闭右边管道，左边只负责发送数据
    for i in seq:
        left.send(i)
        # time.sleep(1)
    else:
        left.close()

if __name__ == '__main__':
    pass
    left,right = multiprocessing.Pipe()

    c1 = multiprocessing.Process(target=consumer,args=((left,right),'c1'))
    c1.start()

    seq = (i for i in range(10))
    producer(seq,(left,right))
    right.close()				# 关闭管道
    left.close()				# 关闭管道
    c1.join()
    print("主进程 over")

c1 收到包子:0
c1 收到包子:1
c1 收到包子:2
c1 收到包子:3
c1 收到包子:4
c1 收到包子:5
c1 收到包子:6
c1 收到包子:7
c1 收到包子:8
c1 收到包子:9
主进程 over

**注意：生产者和消费者都没有使用管道的某个端点，就应该将其关闭，如在生产者中关闭管道的右端，在消费者中关闭管道的左端。如果忘记执行这些步骤，程序可能再消费者中的recv()操作上挂起。管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生产EOFError异常。因此在生产者中关闭管道不会有任何效果，付费消费者中也关闭了相同的管道端点。

管道可以用于双向通信，利用通常在客户端/服务器中使用的请求／响应模型或远程过程调用，就可以使用管道编写与进程交互的程序,如下：

from multiprocessing import Process,Pipe

import time,os
def adder(p,name):
    server,client=p
    client.close()
    while True:
        try:
            x,y=server.recv()
        except EOFError:
            server.close()
            break
        res=x+y
        server.send(res)
    print('server done')
if __name__ == '__main__':
    server,client=Pipe()

    c1=Process(target=adder,args=((server,client),'c1'))
    c1.start()

    server.close()

    client.send((10,20))
    print(client.recv())
    client.close()

    c1.join()
    print('主进程')
#注意：send()和recv()方法使用pickle模块对对象进行序列化