首页 > > 网络编程 > 其它 >

python中的并发编程

2018-06-17 23:33:18来源：未知阅读 ()

一.多进程

1.1 multiprocessing模块和Process类

　　multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。

　　multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

from multiprocessing import Process 用来创建子进程

Process类的参数：

group   参数未使用，值始终为None
target   表示调用对象，即子进程要执行的任务
args   表示调用对象的位置参数元组，args=(1,2)
kwargs   表示调用对象的字典,kwargs={'name':'hantao','age':18}
name   为子进程的名称

Process类的方法：

p.start()   启动进程，并调用该子进程中的p.run()
p.run()    进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate()    强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive()    如果p仍然运行，返回True
p.join([timeout])     主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程

Process类的属性：

p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid
p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束
p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功

1.2 开启子进程

#开启子进程方式一
from multiprocessing import Process
import time,random

def piao(name):
    print('%s is piaoing'%name)
    time.sleep(random.randint(1,3))
    print('%s is over'%name)

if __name__ == '__main__':
    p=Process(target=piao,args=('hantao',))
    p.start()
    print('zhu')

#开启子进程方式二
from multiprocessing import Process
import time,random

class MyProcess(Process):
    def __init__(self,name):
        super(MyProcess,self).__init__()
        self.name=name

    def run(self):
        print('%s is piaoing'%self.name)
        time.sleep(random.randint(1,3))
        print('%s is over'%self.name)

if __name__ == '__main__':
    p=MyProcess('hantao')
    p.start()
    print('zhu')

并发的套接字通信

from socket import *
from multiprocessing import Process

def talk(conn,addr):
    while True:
        try:
            data = conn.recv(1024)
            if not data: break
            conn.send(data.upper())
        except Exception:
            break
    conn.close()

def server():
    s = socket()
    s.bind(('127.0.0.1', 8080))
    s.listen(5)
    while True:
        conn,addr=s.accept()
        print(addr)
        p=Process(target=talk,args=(conn,addr))
        p.start()
    s.close()

if __name__ == '__main__':
    server()

服务端

from socket import *

c=socket()
c.connect(('127.0.0.1',8080))

while True:
    msg=input('>>:').strip()
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

c.close()

客户端

join方法

from multiprocessing import Process
import time,random

def piao(name):
    print('%s is piaoing'%name)
    time.sleep(random.randint(1,3))
    print('%s is over'%name)

if __name__ == '__main__':
    p1=Process(target=piao,args=('hantao',))
    p2=Process(target=piao,args=('hanyn',))
    p3=Process(target=piao,args=('hanxiaoning',))
    p1.start()
    p2.start()
    p3.start()
    p1.join()
    p2.join()
    p3.join()
    print('zhu')

1.3 守护进程

from  multiprocessing import Process
import os
import time

def task():
    print('%s is running'%os.getpid())
    time.sleep(3)
    print('%s is done'%os.getpid())

if __name__ == '__main__':
    p=Process(target=task,)
    p.daemon=True                 #必须早p.start()前设置
    p.start()
    time.sleep(1)
    print('zhu')

#什么时候用守护进程：
#首先开子进程的目的就是为了并发执行任务
#如果该任务的执行周期与主进程的执行周期是一致的
#就必须把该任务的进程设置为守护进程

#守护进程到底什么时候死：主进程执行完毕
#主进程什么时候算执行完毕：主进程运行完毕最后一行代码
#主进程什么时候算死掉：主进程执行完毕并且所有非守护进程都死掉，主进程死掉
#主进程执行完毕了，是否意味着主进程会立马死掉：不是

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__':
    p1=Process(target=foo)
    p2=Process(target=bar)
    p1.daemon=True
    p1.start()
    p2.start()
    print('main')

##守护进程不能开子进程##

1.4 互斥锁

from multiprocessing import Process,Lock
import json
import time
import os
import random

def search():
    data=json.load(open('db.txt',encoding='utf-8'))
    print('剩余票数是:%s'%data['count'])

def get():
    data=json.load(open('db.txt',encoding='utf-8'))
    if data['count']>0:
        data['count']-=1
        time.sleep(random.randint(1,3))
        json.dump(data,open('db.txt','w',encoding='utf-8'))
        print('%s购票成功'%os.getpid())

def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()
    # with lock:
    #     search()
    #     get()

if __name__ == '__main__':
    lock=Lock()
    for i in range(10):
        p=Process(target=task,args=(lock,))
        p.start()

1.5 IPC

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

队列

方法：

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()一样

1 q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
2 q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。
3 q.join_thread()：连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

from multiprocessing import Process,Queue

q=Queue(3)         #队列中允许最大项数，没有则无大小限制
q.put({'name':'hantao'})
q.put('lihaile')
q.put(123)

print(q.get())
print(q.get())
print(q.get())

生产者消费者模型

　　生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

from multiprocessing import Process,Queue,JoinableQueue
import time
import random

def producer(name,food,q):
    for i in range(5):
        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))
        q.task_done()

if __name__ == '__main__':
    q=JoinableQueue()
    p1=Process(target=producer,args=('hantao','蚕豆',q))
    p2=Process(target=producer,args=('xiaoxiaojing','花生米',q))
    c1=Process(target=consumer,args=('喵',q))
    c2=Process(target=consumer,args=('汪',q))
    c3=Process(target=consumer,args=('娃',q))
    p1.start()
    p2.start()
    c1.daemon=True
    c2.daemon=True
    c3.daemon=True
    c1.start()
    c2.start()
    c3.start()

    p1.join()
    p2.join()
    q.join()
    print('zhu')

   #JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

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

管道

pass

共享数据

from multiprocessing import Process,Manager

def task(d):
    d['count']-=1

if __name__ == '__main__':
    m=Manager()
    d=m.dict({'count':10})
    p_l=[]
    for i in range(10):
        p=Process(target=task,args=(d,))
        p_l.append(p)
        p.start()
    for p in p_l:
        p.join()

    print(d)

1.6 进程池

利用concurrent.futures import ProcessPoolExecutor实现

from concurrent.futures import ProcessPoolExecutor
import time,os,random

def task(i):
    print('%s is running %s'%(os.getpid(),i))
    time.sleep(random.randint(1,3))
    return i**2

if __name__ == '__main__':
    pool=ProcessPoolExecutor()

    objs=[]
    for i in range(10):
        obj=pool.submit(task,i)
        objs.append(obj)
    pool.shutdown(wait=True)
    print('zhu')
    for obj in objs:
        print(obj.result())

回调机制

from concurrent.futures import ProcessPoolExecutor
import requests
import os


def get(url):
    print('%s get %s'%(os.getpid(),url))
    response=requests.get(url)
    if response.status_code==200:
        return {'url':url,'text':response.text}

def parse(res):
    res=res.result()
    url=res['url']
    text=res['text']
    print('%s parse%s res:%s'%(os.getpid(),url,len(text)))

if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]
    p=ProcessPoolExecutor()
    for url in urls:
        p.submit(get,url).add_done_callback(parse)
    p.shutdown(wait=True)

from multiprocessing import Process,Pool另一种实现方式

二.多线程

threading模块和multiprocess模块的接口基本相同

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

开启线程的两种方式：

from threading import Thread
import os
import time
import random

def task():
    print('%s is runing'%os.getpid())
    time.sleep(random.randint(1,3))
    print('%s is done'%os.getpid())

if __name__ == '__main__':
    t=Thread(target=task,)
    t.start()
    print('主线程')

方式一

from threading import Thread
import os
import time
import random

class Mythead(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print('%s is running'%os.getpid())
        time.sleep(random.randint(1,3))
        print('%s is done'%os.getpid())

if __name__ == '__main__':
    t=Mythead('线程1')
    t.start()
    print('主线程')

方式二

互斥锁

from threading import Thread,Lock
import time

n=100
def task():
    global n
    lock.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()

if __name__ == '__main__':
    lock=Lock()
    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('zhu',n)

守护线程

from  threading import Thread
import os
import time

def task():
    print('%s is running'%os.getpid())
    time.sleep(3)
    print('%s is done'%os.getpid())

if __name__ == '__main__':
    p=Thread(target=task,)
    p.daemon=True                 
    p.start()
    print('zhu')

信号量

from threading import Thread,Semaphore,current_thread
import time,random

sm=Semaphore(5)
def task():
    with sm:
        print('%s 正在上厕所'%current_thread().getName())
        time.sleep(random.randint(1,3))

if __name__ == '__main__':
    for i in range(20):
        t=Thread(target=task)
        t.start()

线程池

from threading import Thread,Semaphore,current_thread
from concurrent.futures import ThreadPoolExecutor
import time,random

sm=Semaphore(5)
def task():
    print('%s 正在上厕所'%current_thread().getName())
    time.sleep(random.randint(1,3))

if __name__ == '__main__':
    t=ThreadPoolExecutor(5)
    for i in range(20):
        t.submit(task)
    t.shutdown(wait=True)

定时器

from threading import Timer

def hello(id):
    print('hello world',id)

t=Timer(2,hello,args=(30,))
t.start()

线程queue

import queue

q=queue.Queue()
q.put('1')
q.put(2)
q.put(3)

print(q.get())
print(q.get())
print(q.get())

import queue

q=queue.LifoQueue()           #堆栈
q.put('1')
q.put(2)
q.put(3)

print(q.get())
print(q.get())
print(q.get())

import queue
q=queue.PriorityQueue()         #优先级队列
q.put((10,'ha'))
q.put((-2,'nt'))
q.put((5,'ao'))
print(q.get())
print(q.get())
print(q.get())

Python GIL(Global Interpreter Lock)

三. 协程

　　协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。

import time
def consumer():
    '''任务1:接收数据,处理数据'''
    while True:
        x=yield

def producer():
    '''任务2:生产数据'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)

if __name__ == '__main__':

    start=time.time()
    #基于yield保存状态,实现两个任务直接来回切换,即并发的效果
    #PS:如果每个任务中都加上打印,那么明显地看到两个任务的打印是你一次我一次,即并发执行的.
    producer()

    stop=time.time()
    print(stop-start)

利用yield实现协程，并发执行,但是遇到io就阻塞,不会切到该线程

from greenlet import greenlet

def eat(name):
    print('%s eat 1'%name)
    g2.switch('lijing')
    print('%s eat 2'%name)
    g2.switch()

def play(name):
    print('%s play 1'%name)
    g1.switch()
    print('%s play 2'%name)

g1=greenlet(eat)
g2=greenlet(play)
#可以在第一次switch时传入参数，以后都不需要
g1.switch('hantao')

Greenlet实现协程，无法检测io阻塞

Gevent

　　Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。

#用法
g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的

g2=gevent.spawn(func2)

g1.join() #等待g1结束

g2.join() #等待g2结束

#或者上述两步合作一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

　　time.sleep()或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了

from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面

from gevent import monkey;monkey.patch_all()
import gevent
import time

def eat(name):
    print('%s eat 1'%name)
    time.sleep(2)
    print('%s eat 2'%name)
    return 2

def play(name):
    print('%s play 1'%name)
    time.sleep(3)
    print('%s play 2'%name)
    return 3

g1=gevent.spawn(eat,'hantao')
g2=gevent.spawn(play,'lijing')
g1.join()
g2.join()
print(g1.value,g2.value)

from gevent import spawn,joinall,monkey;monkey.patch_all()

import time
def task(pid):
    """
    Some non-deterministic task
    """
    time.sleep(0.5)
    print('Task %s done' % pid)


def synchronous():
    for i in range(10):
        task(i)

def asynchronous():
    g_l=[spawn(task,i) for i in range(10)]
    joinall(g_l)

if __name__ == '__main__':
    print('Synchronous:')
    synchronous()

    print('Asynchronous:')
    asynchronous()
#上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。 初始化的greenlet列表存放在数组threads中，此数组被传给gevent.joinall 函数，后者阻塞当前流程，并执行所有给定的greenlet。执行流程只会在 所有greenlet执行完后才会继续向下走。