In-depth analysis of multi-process, multi-thread and coroutine in python

History of processes and threads

We all know that computers are composed of hardware and software. The CPU in the hardware is the core of the computer and undertakes all the tasks of the computer. The operating system is software that runs on the hardware and is the manager of the computer. It is responsible for the management and allocation of resources and the scheduling of tasks. A program is a software with a certain function that runs on the system, such as a browser, a music player, etc. Every time a program is executed, a certain function will be completed. For example, a browser helps us open a web page. In order to ensure its independence, a special data structure is needed to manage and control the execution of the program - the process control block. A process is a dynamic execution of a program on a data set. A process generally consists of three parts: program, data set, and process control block. The program we write is used to describe what functions the process needs to complete and how to complete it; the data set is the resources that the program needs to use during execution; the process control block is used to record the external characteristics of the process, describe the execution change process of the process, and the system It can be used to control and manage processes. It is the only sign that the system senses the existence of a process.

In early operating systems, computers had only one core, the smallest unit for process execution. Task scheduling used a preemptive method of time slice rotation for process scheduling. Each process has its own independent memory to ensure the isolation of the memory address spaces between processes. With the development of computer technology, processes have appeared many disadvantages. First, the cost of creating, canceling, and switching processes is relatively high. Second, due to the Symmetrical Multi-Processing (Symmetrical Multi-Processing), also called SMP, refers to A group of processors (multiple CPUs) are assembled on one computer, and the emergence of shared memory subsystems and bus structures between CPUs can satisfy multiple running units, but the parallel overhead of multiple processes is too large. At this time, the concept of threads was introduced. A thread is also called a lightweight process. It is a basic CPU execution unit and the smallest unit in the program execution process. It is composed of a thread ID, a program counter, a register set and a stack. The introduction of threads reduces the overhead of concurrent execution of programs and improves the concurrency performance of the operating system. Threads do not have their own system resources, only the resources that are essential during runtime. But threads can share other resources owned by the process with other threads that belong to the same process.

The relationship between processes and threads

Threads belong to processes. Threads run in the process space. Threads generated by the same process share the same memory space. When the process exits, the threads generated by the process will be forced to exit and be cleared. A thread can share all the resources owned by the process with other threads belonging to the same process, but it itself basically does not own system resources, only a little bit of information essential for running (such as a program counter, a set of registers, and a stack).

python thread

Threading is used to provide thread-related operations. A thread is the smallest unit of work in an application.

1. Threading module

The threading module is built on top of the _thread module. The thread module handles and controls threads in a low-level, primitive way, while the threading module provides a more convenient API to handle threads by secondary encapsulation of threads.

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

#!/usr/bin/env python
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)<br>输出：<br>　　3.1415927<br>　　[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)<br><br>输出： 
　　{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()

map(func, iterable[, chunksize])¶
map_async(func, iterable[, chunksize[, callback[, error_callback]]])¶
imap(func, iterable[, chunksize])¶
imap_unordered(func, iterable[, chunksize])
starmap(func, iterable[, chunksize])¶
starmap_async(func, iterable[, chunksize[, callback[, error_back]]])

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.python.org/'),
  gevent.spawn(func, 'https://www.yahoo.com/'),
  gevent.spawn(func, 'https://github.com/'),
])