Pythonのプロセス、スレッド、コルーチンの詳細な説明

import threading
import time
   
def worker(num):
    """
    thread worker function
    :return:
    """
    time.sleep(1)
    print("The num is  %d" % num)
    return
   
for i in range(20):
    t = threading.Thread(target=worker,args=(i,)，name=“t.%d” % i)
    t.start()

import threading
import time
   
globals_num = 0
   
lock = threading.RLock()
   
def Func():
    lock.acquire()  # 获得锁
    global globals_num
    globals_num += 1
    time.sleep(1)
    print(globals_num)
    lock.release()  # 释放锁
   
for i in range(10):
    t = threading.Thread(target=Func)
    t.start()

import threading
lock = threading.Lock()    #Lock对象
lock.acquire()
lock.acquire()  #产生了死琐。
lock.release()
lock.release()　
import threading
rLock = threading.RLock()  #RLock对象
rLock.acquire()
rLock.acquire()    #在同一线程内，程序不会堵塞。
rLock.release()
rLock.release()

import threading
   
def do(event):
    print('start')
    event.wait()
    print('execute')
   
event_obj = threading.Event()
for i in range(10):
    t = threading.Thread(target=do, args=(event_obj,))
    t.start()
   
event_obj.clear()
inp = input('input:')
if inp == 'true':
    event_obj.set()

import threading
import time
def consumer(cond):
    with cond:
        print("consumer before wait")
        cond.wait()
        print("consumer after wait")
   
def producer(cond):
    with cond:
        print("producer before notifyAll")
        cond.notifyAll()
        print("producer after notifyAll")
   
condition = threading.Condition()
c1 = threading.Thread(name="c1", target=consumer, args=(condition,))
c2 = threading.Thread(name="c2", target=consumer, args=(condition,))
   
p = threading.Thread(name="p", target=producer, args=(condition,))
   
c1.start()
time.sleep(2)
c2.start()
time.sleep(2)
p.start()

import queue
  
q = queue.Queue(maxsize=0)  # 构造一个先进显出队列，maxsize指定队列长度，为0 时，表示队列长度无限制。
  
q.join()    # 等到队列为kong的时候，在执行别的操作
q.qsize()   # 返回队列的大小 （不可靠）
q.empty()   # 当队列为空的时候，返回True 否则返回False （不可靠）
q.full()    # 当队列满的时候，返回True，否则返回False （不可靠）
q.put(item, block=True, timeout=None) #  将item放入Queue尾部，item必须存在，可以参数block默认为True,表示当队列满时，会等待队列给出可用位置，
　　　　　　　　　　　　　　　　　　　　　　　　 为False时为非阻塞，此时如果队列已满，会引发queue.Full 异常。 可选参数timeout，表示 会阻塞设置的时间，过后，
　　　　　　　　　　　　　　　　　　　　　　　　  如果队列无法给出放入item的位置，则引发 queue.Full 异常
q.get(block=True, timeout=None) #   移除并返回队列头部的一个值，可选参数block默认为True，表示获取值的时候，如果队列为空，则阻塞，为False时，不阻塞，
　　　　　　　　　　　　　　　　　　　　　　若此时队列为空，则引发 queue.Empty异常。 可选参数timeout，表示会阻塞设置的时候，过后，如果队列为空，则引发Empty异常。
q.put_nowait(item) #   等效于 put(item,block=False)
q.get_nowait() #    等效于 get(item,block=False)

import Queue
import threading
message = Queue.Queue(10)
  
  
def producer(i):
    while True:
        message.put(i)
  
  
def consumer(i):
    while True:
        msg = message.get()
  
  
for i in range(12):
    t = threading.Thread(target=producer, args=(i,))
    t.start()
  
for i in range(10):
    t = threading.Thread(target=consumer, args=(i,))
    t.start()

# 简单往队列中传输线程数
import threading
import time
import queue
class Threadingpool():
    def __init__(self,max_num = 10):
        self.queue = queue.Queue(max_num)
        for i in range(max_num):
            self.queue.put(threading.Thread)
    def getthreading(self):
        return self.queue.get()
    def addthreading(self):
        self.queue.put(threading.Thread)
def func(p,i):
    time.sleep(1)
    print(i)
    p.addthreading()
if __name__ == "__main__":
    p = Threadingpool()
    for i in range(20):
        thread = p.getthreading()
        t = thread(target = func, args = (p,i))
        t.start()

#往队列中无限添加任务
import queue
import threading
import contextlib
import time
StopEvent = object()
class ThreadPool(object):
    def __init__(self, max_num):
        self.q = queue.Queue()
        self.max_num = max_num
        self.terminal = False
        self.generate_list = []
        self.free_list = []
    def run(self, func, args, callback=None):
        """
        线程池执行一个任务
        :param func: 任务函数
        :param args: 任务函数所需参数
        :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
        :return: 如果线程池已经终止，则返回True否则None
        """
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            self.generate_thread()
        w = (func, args, callback,)
        self.q.put(w)
    def generate_thread(self):
        """
        创建一个线程
        """
        t = threading.Thread(target=self.call)
        t.start()
    def call(self):
        """
        循环去获取任务函数并执行任务函数
        """
        current_thread = threading.currentThread
        self.generate_list.append(current_thread)
        event = self.q.get()  # 获取线程
        while event != StopEvent:   # 判断获取的线程数不等于全局变量
            func, arguments, callback = event   # 拆分元祖，获得执行函数，参数，回调函数
            try:
                result = func(*arguments)   # 执行函数
                status = True
            except Exception as e:    # 函数执行失败
                status = False
                result = e
            if callback is not None:
                try:
                    callback(status, result)
                except Exception as e:
                    pass
            # self.free_list.append(current_thread)
            # event = self.q.get()
            # self.free_list.remove(current_thread)
            with self.work_state():
                event = self.q.get()
        else:
            self.generate_list.remove(current_thread)
    def close(self):
        """
        关闭线程，给传输全局非元祖的变量来进行关闭
        :return:
        """
        for i in range(len(self.generate_list)):
            self.q.put(StopEvent)
    def terminate(self):
        """
        突然关闭线程
        :return:
        """
        self.terminal = True
        while self.generate_list:
            self.q.put(StopEvent)
        self.q.empty()
    @contextlib.contextmanager
    def work_state(self):
        self.free_list.append(threading.currentThread)
        try:
            yield
        finally:
            self.free_list.remove(threading.currentThread)
def work(i):
    print(i)
    return i +1 # 返回给回调函数
def callback(ret):
    print(ret)
pool = ThreadPool(10)
for item in range(50):
    pool.run(func=work, args=(item,),callback=callback)
pool.terminate()
# pool.close()

from multiprocessing import Process
  
def func(name):
    print('hello', name)
  
  
if __name__ == "__main__":
    p = Process(target=func,args=('zhangyanlin',))
    p.start()
    p.join()  # 等待进程执行完毕

from multiprocessing import Array,Value,Process
  
def func(a,b):
    a.value = 3.333333333333333
    for i in range(len(b)):
        b[i] = -b[i]
  
  
if __name__ == "__main__":
    num = Value('d',0.0)
    arr = Array('i',range(11))
  
  
    c = Process(target=func,args=(num,arr))
    d= Process(target=func,args=(num,arr))
    c.start()
    d.start()
    c.join()
    d.join()
  
    print(num.value)
    for i in arr:
        print(i)

3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

from multiprocessing import Process,Manager
def f(d,l):
    d["name"] = "zhangyanlin"
    d["age"] = 18
    d["Job"] = "pythoner"
    l.reverse()
  
if __name__ == "__main__":
    with Manager() as man:
        d = man.dict()
        l = man.list(range(10))
  
        p = Process(target=f,args=(d,l))
        p.start()
        p.join()
  
        print(d)
        print(l)

{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

#apply
from  multiprocessing import Pool
import time
  
def f1(i):
    time.sleep(0.5)
    print(i)
    return i + 100
  
if __name__ == "__main__":
    pool = Pool(5)
    for i in range(1,31):
        pool.apply(func=f1,args=(i,))
  
#apply_async
def f1(i):
    time.sleep(0.5)
    print(i)
    return i + 100
def f2(arg):
    print(arg)
  
if __name__ == "__main__":
    pool = Pool(5)
    for i in range(1,31):
        pool.apply_async(func=f1,args=(i,),callback=f2)
    pool.close()
    pool.join()

import asyncio
   
async def cor1():
    print("COR1 start")
    await cor2()
    print("COR1 end")
   
async def cor2():
    print("COR2")
   
loop = asyncio.get_event_loop()
loop.run_until_complete(cor1())
loop.close()

import greenlet
def fun1():
    print("12")
    gr2.switch()
    print("56")
    gr2.switch()
  
def fun2():
    print("34")
    gr1.switch()
    print("78")
  
  
gr1 = greenlet.greenlet(fun1)
gr2 = greenlet.greenlet(fun2)
gr1.switch()

pip3 install --upgrade pip3
pip3 install gevent

import gevent
  
def fun1():
    print("www.baidu.com")   # 第一步
    gevent.sleep(0)
    print("end the baidu.com")  # 第三步
  
def fun2():
    print("www.zhihu.com")   # 第二步
    gevent.sleep(0)
    print("end th zhihu.com")  # 第四步
  
gevent.joinall([
    gevent.spawn(fun1),
    gevent.spawn(fun2),
])

import gevent
import requests
def func(url):
    print("get: %s"%url)
    gevent.sleep(0)
    date =requests.get(url)
    ret = date.text
    print(url,len(ret))
gevent.joinall([
    gevent.spawn(func, 'https://www.pythontab.com/'),
    gevent.spawn(func, 'https://www.yahoo.com/'),
    gevent.spawn(func, 'https://github.com/'),
])