您现在的位置是：首页 > 文章详情

python-进程之间通信、多线程介绍

日期：2018-09-05点击：324收藏

一、进程之间通信

进程的任务有三种状态：运行，就绪，阻塞。

加锁可以让多个进程修改同一块数据时，同一时间只能由一个任务可以进行修改，即串行的修改。牺牲了速度，保证了数据安全。

虽然可以使用文件共享数据实现进程间的通信，但是效率太低，还需要自己加锁处理。为了解决这些问题，便使用到了multiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道

1.队列和管道都是将数据存放于内存中

2.队列是基于管道+锁的机制实现的。

我们应该尽量避免使用共享数据，多使用队列。

队列：

创建队列的类：

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

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

q = Queue() q.put（） 括号里可以是任意类型，不能是大数据

主要方法：

1 q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。没有参数时，q.put的个数大于队列数时，会一直阻塞住。 2 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常。没有参数时，q.get的个数大于队列数时，会一直阻塞住。 3.q.put_nowait()等价于q.put(block=False)队列满时再存也会抛异常 q.get_nowait()等价于q.get(block=False)队列为空取不出时会抛异常

from multiprocessing import Queue q=Queue(3) #1.不应该放大数据 2.可以任意类型 q.put(['first',]) q.put({'x':2,}) q.put(3) # q.put(6)#阻塞在存放的时候，当有内容被取出时才推出阻塞模式 print(q.get()) print(q.get()) print(q.get()) # print(q.get())#阻塞在取的时候，当有新内容添加进去才退出阻塞状态

test

生产者消费者模型：

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。通俗的讲就是当程序中出现明显的两类任务，一类负责生产数据，一类负责处理数据，就可以引入这个模型来实现生产者与消费者的解耦合，平衡生产力与消费能力，从而提升效率。

初级版本：

import time import random from multiprocessing import Process,Queue def producer(name,food,q): for i in range(4): res = '%s%s'%(food,i) time.sleep(random.randint(1,3)) q.put(res) print('厨师[%s]生产了%s'%(name,res)) def consumer(name,q): while True: res = q.get() # if res ==None:break time.sleep(random.randint(1,3)) print('吃货%s吃了%s'%(name,res)) if __name__ == '__main__': q = Queue() #生产者 p1 = Process(target=producer,args=('egon','包子',q)) p2 = Process(target=producer,args=('lxx','豆浆',q)) p3 = Process(target=producer,args=('cw','油条',q)) #消费者 c1 = Process(target=consumer,args=('alex',q)) c2 = Process(target=consumer,args=('yxf',q)) p1.start() p2.start() p3.start() c1.start() c2.start() print('主')

test1

这个时候就有一个问题，主进程一直在阻塞状态，永远不会结束。这是因为消费者在取空以后还一直在原地阻塞着。

解决方法：让生产者在生产完毕后再往队列中发送一个结束信号，这样消费者在收到结束信号后跳出循环

版本二：

import time import random from multiprocessing import Process,Queue def producer(name,food,q): for i in range(4): res = '%s%s'%(food,i) time.sleep(random.randint(1,3)) q.put(res) print('厨师[%s]生产了%s'%(name,res)) def consumer(name,q): while True: res = q.get() if res ==None:break time.sleep(random.randint(1,3)) print('吃货%s吃了%s'%(name,res)) if __name__ == '__main__': q = Queue() #生产者 p1 = Process(target=producer,args=('egon','包子',q)) p2 = Process(target=producer,args=('lxx','豆浆',q)) p3 = Process(target=producer,args=('cw','油条',q)) #消费者 c1 = Process(target=consumer,args=('alex',q)) c2 = Process(target=consumer,args=('yxf',q)) p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() q.put(None) q.put(None) q.put(None) print('主')

test2

但是有几个生产者我们就要发送几个None，这不够优雅，于是我们导入了JoinableQueue模块

#JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。 #参数介绍：  maxsize是队列中允许最大项数，省略则无大小限制。 #方法介绍：  JoinableQueue的实例p除了与Queue对象相同的方法之外还具有： q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常 q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

版本三：

import time import random from multiprocessing import Process,JoinableQueue def producer(name,food,q): for i in range(4): res = '%s%s'%(food,i) time.sleep(random.randint(1,3)) q.put(res) print('厨师[%s]生产了%s'%(name,res)) def consumer(name,q): while True: res = q.get() time.sleep(random.randint(1,3)) print('吃货%s吃了%s'%(name,res)) q.task_done() if __name__ == '__main__': q = JoinableQueue() #生产者 p1 = Process(target=producer,args=('egon','包子',q)) p2 = Process(target=producer,args=('lxx','豆浆',q)) p3 = Process(target=producer,args=('cw','油条',q)) #消费者 c1 = Process(target=consumer,args=('alex',q)) c2 = Process(target=consumer,args=('yxf',q)) c1.daemon=True c2.daemon=True p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() q.join()#主进程等q结束，即q内数据被取干净了。 print('主')

View Code

二、线程理论

进程是资源单位，由若干线程组成的，一个进程至少有一个线程，线程是操作系统直接支持的执行单元。一个进程相当于一块空间，空间内有若干的线程，就好比进程是一个车间，线程就是车间的流水线。同一个进程之间的线程是资源共享的。进程是程序执行的过程，那么线程就是程序执行的代码。

线程相比进程的优点：1.同一进程下，多个线程共享进程内的资源

2.创建线程的开销远远小于进程

三、线程的使用

（一）.创建线程的两种方式（和进程类似）

from threading import Thread import time def task(name): print('%s is running'%name) time.sleep(2) print('%s is done'%name) if __name__ == '__main__': t = Thread(target=task,args=('进程一',)) t.start() print('主')

方式一

直接导入threading模块

from threading import Thread import time class Mythead(Thread): def __init__(self,name): super().__init__() self.name = name def run(self): print('%s is running' % self.name) time.sleep(0.1) print('%s is done' % self.name) if __name__ == '__main__': t = Mythead('线程一') t.start() print('主')

方式二

继承Therad类重写run方法

（二）.线程特性介绍

os模块下的getpid（）线程也适用 current_thread#查看当前线程 active_count#查看线程的活跃个数（线程已结束的不是活跃状态） current_thread().name查看线程名字 略

（三）.守护线程

无论是进程还是线程都遵循某线程/进程等待主线程/进程运行完毕后才会被销毁。强调：运行完毕并非终止运行

#1.对主进程来说，运行完毕指的是主进程代码运行完毕。  主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束， #2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕。 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

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=Process(target=foo) # t2=Process(target=bar) t1.daemon=True t1.start() t2.start() print("main-------")

进程与线程的区别

（四）.线程的互斥锁

不加锁的情况

from threading import Thread,Lock import time mutex=Lock() n=100 def task(): global n # mutex.acquire() temp=n time.sleep(0.1) n=temp-1 # mutex.release() if __name__ == '__main__': t_l=[] for i in range(100): t=Thread(target=task) t_l.append(t) t.start() for t in t_l: t.join() print(n)

View Code

结果永远为99，因为线程速度太快了，并发的时候都读到了n=100，所以需要给他加上锁

加上锁的情况

from threading import Thread,Lock import time mutex=Lock() n=100 def task(): global n mutex.acquire() temp=n time.sleep(0.1) n=temp-1 mutex.release() if __name__ == '__main__': t_l=[] for i in range(100): t=Thread(target=task) t_l.append(t) t.start() for t in t_l: t.join() print(n)