python-多进程、线程、协程

进程线程协程分别是什么？

        进程是具有独立功能程序关于某个数据集合上的一次运行活动，进程是系统进行资源分配和调度的一个独立单位。

        线程是进程的一个实体，是cpu和分派的基本单位，它比进程更小的的独立运行的基本单位，线程自己基本不拥有系统资源，但它可与用属于一个进程的其他线程共享进程所拥有的全部资源。

        协程其实可以认为是比线程更小的执行单元，协程拥有自己的寄存器上下文和栈。我们把一个线程中的一个个函数叫做子程序，那么子程序在执行过程中可以中断去执行别的子程序，别的子程序也可以中断回来继续执行之前的子程序，这就是协程。

进程

        同一个program程序可以同时存在多个process进程；

        一个cpu一次只能执行一个 process，每一个process都是相互独立的；

        同一个process下的Thread共享process的资源，如内存、变量等，不同的进程不能共享；

"""
进程不阻塞，会先执行主进程，再执行子进程
进程阻塞，join()   会等待子进程执行完，再往下执行主进程
进程守护，setDaemon(True) 主进程结束后，子进程也会随之结束；  进程阻塞和进程守护同时存在时以阻塞为主
进程通信，Queue()  采用put和get方法
进程数据共享 Manager()
"""

from multiprocessing import Process
import time
#进程阻塞
def worker(q):
    time.sleep(1)
    print(f'Process {q}')

if __name__ == '__main__':
    print('Start')
    p = Process(target=worker, args=('Process-1',))
    p.start()
    #p.join()   #阻塞
    print('Finished')


from multiprocessing import Process
#守护进程

def worker(q):
    print(f'Process {q}')

if __name__ == '__main__':
    print('Start')
    p1 = Process(daemon=True,target=worker, args=('Process-1',)) #守护进程，主代码结束守护进程随之结束
    p1.start()
    print('Finished')


from multiprocessing import Process
import os
#多进程阻塞

def worker(q):
    print(f'Process {q}  son:{os.getpid()}  father:{os.getppid()}')

if __name__ == '__main__':
    print('Start')
    p1 = Process(target=worker, args=('Process-1',))
    p2 = Process(target=worker, args=('Process-2',))
    p1.start()
    p2.start()
    print(f'Finished   son:{os.getpid()}   pycharm:{os.getppid()}')


from multiprocessing import Process, Queue
#进程通信

def p_put(q):
    q.put('hello')
    print('put',q.qsize())

def p_get(q):
    print(q.get())
    print('get',q.qsize())

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=p_put, args=(q, ))
    p2 = Process(target=p_get, args=(q, ))
    p1.start()
    p2.start()


from multiprocessing import Process,Manager
#进程数据共享

def func(mydict, mylist):
    mydict["name"] = "shuzf"  # 子进程改变dict,主进程跟着改变
    mylist.append(11)            # 子进程改变List,主进程跟着改变


if __name__ == "__main__":
    with Manager() as MG:
        mydict = MG.dict()          # 主进程与子进程共享这个字典
        mylist = MG.list(range(2))  # 主进程与子进程共享这个List

        p = Process(target=func, args=(mydict, mylist))
        p.start()
        p.join()

        print(mylist)
        print(mydict)

线程

　　同一个 process进程可以同时存在多个thread线程

　　process本身不是基本执行单位，它是thread（基本执行单位）的容器;

　　同一个process下的Thread共享process的资源，如内存、变量等，不同的进程不能共享;

　　在多线程（Multi threading）中，两个线程如果同时存取或改变全局变量（Golbal Variable），则可能发生同步（Synchronization）的问题。如果线程之间互抢资源，则可能产生死锁（Dead Lock）。

"""
线程
线程不阻塞，会先执行主线程，再执行子线程
线程阻塞，join()  会等待子线程执行完，再往下执行主线程
守护线程，setDaemon(True) 主线程结束后，子线程也会随之结束；  线程阻塞和守护线程同时存在时以阻塞为主
线程锁，Lock()  RLock() Event()
"""
from threading import Thread
import time
#单线程

def work(q):
    time.sleep(1)
    print(f'Thread {q}')

if __name__ == '__main__':
    print("Starting Main Thread")
    t=Thread(target=work,args=("Thread-1",))
    t.start()
    #t.join()
    print("Exiting Main Thread")



from threading import Thread
import time
#守护线程

def work(q):
    time.sleep(1)
    print(f'Thread {q}')

if __name__ == '__main__':
    print("Starting Main Thread")
    t=Thread(target=work,args=("Thread-1",))
    t.setDaemon(True)
    t.start()
    t.join()
    print("Exiting Main Thread")


from threading import Thread
import time
#多线程

def work(q):
    time.sleep(1)
    print(f'Thread {q}')

if __name__ == '__main__':
    print("Starting Main Thread")
    t1=Thread(target=work,args=("Thread-1",))
    t2=Thread(target=work,args=("Thread-2",))
    t1.start()
    t2.start()
    # t1.join()
    # t2.join()
    print("Exiting Main Thread")


from threading import Thread
#多线程方式2
class myThread(Thread):
    def __init__(self,threadName):
        super().__init__()
        self.threadName=threadName

    def run(self):
        print(f'Thread runing {self.threadName}')

if __name__ == "__main__":
    threads = [myThread("Thread-1"),myThread("Thread-2"),myThread("Thread-3")]
    for i in threads: i.start()     # 开起子线程
    for i in threads: i.join()      # join()子线程执行完，主线程才恢复到运行
    print ("Exiting Main Thread")


from threading import Thread
import threading,time
#线程锁（互斥锁、递归锁）

def work(q):
    print(f"Thread starting {q}")
    threadLock.acquire()  # 获得锁
    time.sleep(1)
    print(f'Thread runing {q}')
    threadLock.release()  # 释放锁
    print(f"Thread exiting {q}")

if __name__ == '__main__':
    print("Starting Main Thread")
    t1=Thread(target=work,args=("Thread-1",))
    t2=Thread(target=work,args=("Thread-2",))
    threadLock = threading.Lock()   # 创建互斥锁（线程1进入锁后，只有释放锁才能执行线程2）
    #threadLock = threading.RLock()  # 创建递归锁（线程1进入锁后，只有释放锁才能执行线程2）
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("Exiting Main Thread")



from threading import Thread
import threading,time
#线程锁（信号量）

def work(q):
    print(f"Thread starting {q}")
    if threadLock.acquire():  # 获得锁
        time.sleep(1)
        print(f'Thread runing {q}')
        threadLock.release()  # 释放锁
        print(f"Thread exiting {q}")

if __name__ == '__main__':
    print("Starting Main Thread")
    t1=Thread(target=work,args=("Thread-1",))
    t2=Thread(target=work,args=("Thread-2",))
    t3=Thread(target=work,args=("Thread-3",))
    threadLock  = threading.BoundedSemaphore(2)  #创建信号量（该锁用于限制线程的并发量）
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
    print("Exiting Main Thread")


import time
from threading import Thread,Event
#线程通信（Event）

def foo(event):
    print('wait server...')
    event.wait()    #括号里可以带数字执行，数字表示等待的秒数，不带数字表示一直阻塞状态
    print('connect to server')
def start(event):
    print('start server successful')
    time.sleep(1)
    event.set()     #默认为False，set一次表示True，所以子线程里的foo函数解除阻塞状态继续执行

if __name__ == '__main__':
    event = Event()
    t1=Thread(target=foo,args=(event,))
    t2=Thread(target=start,args=(event,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()


import queue
from threading import Thread
#queue 线程队列

def put(q):
    q.put('hello')
    print('put',q.qsize())

def get(q):
    print(q.get())
    print('get',q.qsize())

if __name__ == '__main__':
    q = queue.Queue()
    t1=Thread(target=put,args=(q,))
    t2=Thread(target=get,args=(q,))
    t1.start()
    t2.start()

协程

　　协程，又称微线程，英文名Coroutine。协程是一种用户态的轻量级线程。

　　我们把一个线程中的一个个函数叫做子程序，那么子程序在执行过程中可以中断去执行别的子程序；别的子程序也可以中断回来继续执行之前的子程序，这就是协程。

　　协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。

import grequests
import requests
import gevent
import asyncio,aiohttp,time,os
from functools import wraps #@wraps可以保证装饰器修饰的函数的name的值保持不变

url1 = "http://www.baidu.com"
url2 ="http://42.192.162.8:9001/api/time"

def runTime(func):
    """
    修饰函数（统计函数运行时间）-无参数、有参数
    @param func: 目标函数
    @return:
    """
    @wraps(func)  #@wraps不改变使用装饰器原有函数的结构(如name, doc)
    def inner(*args, **kwargs):
        s = time.time()
        func(*args, **kwargs)
        print(f'--> {func.__name__} RUN TIME: <{(time.time() - s)}> ')
        #print('--> task  (pid={}) is running '.format(os.getpid()))
    return inner


@runTime
def test_requests(num):
    urls = [url1 for ị in range(num)]
    req = [requests.get(url=url, timeout=5) for url in urls]
    print(req)


@runTime
def test_grequests(num):
    urls = [url1 for ị in range(num)]
    req = [grequests.request("GET", url, timeout=5) for url in urls]
    resp = grequests.map(req, size=100)     #此处map的requests参数是装有实例化请求对象的列表,其返回值也是列表， size参数可以控制并发的数量
    #for res in resp: print(res) #print(res.text)
    print(resp)



async def fetch(url,sem):
    async with sem:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=5) as res:
                 pass   #print(res.status)

@runTime
def test_asyncio(num):
    sem =asyncio.Semaphore(100)  #asyncio控制并发数
    tasks = [fetch(url1,sem) for i in range(num)]
    loop = asyncio.get_event_loop()
    resp = loop.run_until_complete(asyncio.wait(tasks))  #loop.run_until_complete将task放到loop中，进行事件循环,这里必须传入的是一个list
    loop.close()
    print(resp)


def func(url):
    res = requests.get(url=url,timeout=5)
    #print(res)

@runTime
def test_gevent(num):
    tasks= [gevent.spawn(func,url1) for i in range(num)]
    resp = gevent.joinall(tasks)
    print(resp)


if __name__ == "__main__":
    #test_requests(10)
    #test_grequests(10)
    #test_asyncio(10)
    test_gevent(10)

用yield实现协程

"""
通过调用send方法和yield互相切换，实现协程的功能
"""
import time
def consume():
    r=''
    while True:
        n = yield r
        if n is None: return
        print('[consumer] 消费 %s' % n)
        time.sleep(1)
        r = 'well received'

def produce(c:object,n:int=0):
    c.send(None)  #启动生成器，第一次n为None
    while n < 5:
        n = n + 1
        print('[producer] 生产 %s' % n)
        r = c.send(n)  #切换到consume执行
        print('[producer] return: %s' % r)
    c.close()

if __name__=='__main__':
    c = consume()
    produce(c)