python线程、进程和协程详解

server = socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer)

class ThreadingMixIn:
  
 daemon_threads = False
 
 def process_request_thread(self, request, client_address):  
  try:
   self.finish_request(request, client_address)
   self.shutdown_request(request)
  except:
   self.handle_error(request, client_address)
   self.shutdown_request(request)
 
 def process_request(self, request, client_address):
   
  t = threading.Thread(target = self.process_request_thread,
        args = (request, client_address))
  t.daemon = self.daemon_threads
  t.start()

import threading
import time
 
def show(arg):
 time.sleep(1)
 print('thread'+str(arg))
 
for i in range(10):
 t = threading.Thread(target=show, args=(i,))
 t.start()
 
print('main thread stop')

import threading
class MyThreading(threading.Thread):

 def __init__(self,func,arg):
  super(MyThreading,self).__init__()
  self.func = func
  self.arg = arg

 def run(self):
  self.func(self.arg)

def f1(args):
 print(args)

obj = MyThreading(f1, 123)
obj.start()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time
NUM = 0
def show():
 global NUM
 NUM += 1
 name = t.getName()
 time.sleep(1)  # 注意，这行语句的位置很重要，必须在NUM被修改后，否则观察不到脏数据的现象。
 print(name, "执行完毕后，NUM的值为： ", NUM)

for i in range(10):
 t = threading.Thread(target=show)
 t.start()
print('main thread stop')

main thread stop
Thread-1 执行完毕后，NUM的值为： 10
Thread-2 执行完毕后，NUM的值为： 10
Thread-4 执行完毕后，NUM的值为： 10
Thread-9 执行完毕后，NUM的值为： 10
Thread-3 执行完毕后，NUM的值为： 10
Thread-6 执行完毕后，NUM的值为： 10
Thread-8 执行完毕后，NUM的值为： 10
Thread-7 执行完毕后，NUM的值为： 10
Thread-5 执行完毕后，NUM的值为： 10
Thread-10 执行完毕后，NUM的值为： 10

import time
import threading

NUM = 10
def func(lock):
 global NUM
 lock.acquire() # 让锁开始起作用
 NUM -= 1
 time.sleep(1)
 print(NUM)
 lock.release() # 释放锁
 
lock = threading.Lock() # 实例化一个锁对象
for i in range(10):
 t = threading.Thread(target=func, args=(lock,)) # 记得把锁当作参数传递给func参数
 t.start()

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import time
import threading

def run(n):
 semaphore.acquire()
 print("run the thread: %s" % n)
 time.sleep(1)
 semaphore.release()

num = 0
semaphore = threading.BoundedSemaphore(5) # 最多允许5个线程同时运行
for i in range(20):
 t = threading.Thread(target=run, args=(i,))
 t.start()

import threading

def func(e,i):
 print(i)
 e.wait() # 检测当前event是什么状态，如果是红灯，则阻塞，如果是绿灯则继续往下执行。默认是红灯。
 print(i+100)

event = threading.Event()
for i in range(10):
 t = threading.Thread(target=func, args=(event, i))
 t.start()

event.clear() # 主动将状态设置为红灯
inp = input(">")
if inp == "1":
 event.set() # 主动将状态设置为绿灯

import threading

def condition():
 ret = False
 r = input(">")
 if r == "yes":
  ret = True
 return ret

def func(conn, i):
 print(i)
 conn.acquire()
 conn.wait_for(condition) # 这个方法接受一个函数的返回值
 print(i+100)
 conn.release()

c = threading.Condition()
for i in range(10):
 t = threading.Thread(target=func, args=(c, i,))
 t.start()

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import threading

def run(n):
 con.acquire()
 con.wait()
 print("run the thread: %s" %n)
 con.release()

if __name__ == '__main__':

 con = threading.Condition()
 for i in range(10):
  t = threading.Thread(target=run, args=(i,))
  t.start()

 while True:
  inp = input('>')
  if inp == "q":
   break
  # 下面这三行是固定语法
  con.acquire()
  con.notify(int(inp)) # 这个方法接收一个整数，表示让多少个线程通过
  con.release()

from threading import Timer
def hello():
 print("hello, world")
t = Timer(1, hello) # 表示1秒后执行hello函数
t.start() 

import queue
q = queue.Queue(5)
q.put(11)
q.put(22)
q.put(33)

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

def __init__(self, maxsize=0):
self.maxsize = maxsize
self._init(maxsize)

def join(self):
 with self.all_tasks_done:
  while self.unfinished_tasks:
   self.all_tasks_done.wait()

import queue
q = queue.Queue(5)
q.put(11)
q.put(22)
print(q.get())
q.task_done()
print(q.get())
q.task_done()
q.join()

import queue
q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())

q = queue.PriorityQueue()
q.put((1,"alex1"))
q.put((1,"alex2"))
q.put((1,"alex3"))
q.put((3,"alex3"))
print(q.get())

q = queue.deque()
q.append(123)
q.append(333)
q.appendleft(456)
q.pop()
q.popleft()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang
import time
import queue
import threading

q = queue.Queue(10)
def productor(i):
 while True:
  q.put("厨师 %s 做的包子！"%i)
  time.sleep(2)

def consumer(k):
 while True:
  print("顾客 %s 吃了一个 %s"%(k,q.get()))
  time.sleep(1)

for i in range(3):
 t = threading.Thread(target=productor,args=(i,))
 t.start()

for k in range(10):
 v = threading.Thread(target=consumer,args=(k,))
 v.start()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang

import queue
import time
import threading

class MyThreadPool:
 def __init__(self, maxsize=5):
  self.maxsize = maxsize
  self._q = queue.Queue(maxsize)
  for i in range(maxsize):
   self._q.put(threading.Thread)
   
 def get_thread(self):
  return self._q.get()
  
 def add_thread(self):
  self._q.put(threading.Thread)

def task(i, pool):
 print(i)
 time.sleep(1)
 pool.add_thread()

pool = MyThreadPool(5)
for i in range(100):
 t = pool.get_thread()
 obj = t(target=task, args=(i,pool))
 obj.start()

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import queue
import threading
import contextlib
import time

StopEvent = object() # 创建空对象

class ThreadPool(object):

 def __init__(self, max_num, max_task_num = None):
  if max_task_num:
   self.q = queue.Queue(max_task_num)
  else:
   self.q = queue.Queue()
  self.max_num = max_num
  self.cancel = False
  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 self.cancel:
   return
  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)
    success = True
   except Exception as e:
    success = False
    result = None

   if callback is not None:
    try:
     callback(success, result)
    except Exception as e:
     pass

   with self.worker_state(self.free_list, current_thread):
    if self.terminal:
     event = StopEvent
    else:
     event = self.q.get()
  else:

   self.generate_list.remove(current_thread)

 def close(self):
  """
  执行完所有的任务后，所有线程停止
  """
  self.cancel = True
  full_size = len(self.generate_list)
  while full_size:
   self.q.put(StopEvent)
   full_size -= 1

 def terminate(self):
  """
  无论是否还有任务，终止线程
  """
  self.terminal = True

  while self.generate_list:
   self.q.put(StopEvent)

  self.q.empty()

 @contextlib.contextmanager
 def worker_state(self, state_list, worker_thread):
  """
  用于记录线程中正在等待的线程数
  """
  state_list.append(worker_thread)
  try:
   yield
  finally:
   state_list.remove(worker_thread)

# How to use
pool = ThreadPool(5)

def callback(status, result):
 # status, execute action status
 # result, execute action return value
 pass

def action(i):
 print(i)

for i in range(30):
 ret = pool.run(action, (i,), callback)

time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
# pool.close()
# pool.terminate()

from multiprocessing import Process

def foo(i):
 print("This is Process ", i)

if __name__ == '__main__':
 for i in range(5):
  p = Process(target=foo, args=(i,))
  p.start()

from multiprocessing import Process
list_1 = []
def foo(i):
 list_1.append(i)
 print("This is Process ", i," and list_1 is ", list_1)

if __name__ == '__main__':
 for i in range(5):
  p = Process(target=foo, args=(i,))
  p.start()

 print("The end of list_1:", list_1)

from multiprocessing import Process
from multiprocessing import Array

def Foo(i,temp):
 temp[0] += 100
 for item in temp:
  print(i,'----->',item)

if __name__ == '__main__':
 temp = Array('i', [11, 22, 33, 44])
 for i in range(2):
  p = Process(target=Foo, args=(i,temp))
  p.start()

from multiprocessing import Process,Manager

def Foo(i,dic):
 dic[i] = 100+i
 print(dic.values())

if __name__ == '__main__':
 manage = Manager()
 dic = manage.dict()
 for i in range(10):
  p = Process(target=Foo, args=(i,dic))
  p.start()
  p.join()

import multiprocessing
from multiprocessing import Process
from multiprocessing import queues

def foo(i,arg):
 arg.put(i)
 print('The Process is ', i, "and the queue's size is ", arg.qsize())

if __name__ == "__main__":
 li = queues.Queue(20, ctx=multiprocessing)
 for i in range(10):
  p = Process(target=foo, args=(i,li,))
  p.start()

from multiprocessing import Process
from multiprocessing import queues
from multiprocessing import Array
from multiprocessing import RLock, Lock, Event, Condition, Semaphore
import multiprocessing
import time

def foo(i,lis,lc):
 lc.acquire()
 lis[0] = lis[0] - 1
 time.sleep(1)
 print('say hi',lis[0])
 lc.release()

if __name__ == "__main__":
 # li = []
 li = Array('i', 1)
 li[0] = 10
 lock = RLock()
 for i in range(10):
  p = Process(target=foo,args=(i,li,lock))
  p.start()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Pool
import time

def f1(args):
 time.sleep(1)
 print(args)

if __name__ == '__main__':
 p = Pool(5)
 for i in range(30):
  p.apply_async(func=f1, args= (i,))
 p.close()   # 等子进程执行完毕后关闭线程池
 # time.sleep(2)
 # p.terminate()  # 立刻关闭线程池
 p.join()

from greenlet import greenlet

def test1():
 print(12)
 gr2.switch()
 print(34)
 gr2.switch()

def test2():
 print(56)
 gr1.switch()
 print(78)

gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

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

def f(url):
 print('GET: %s' % url)
 resp = requests.get(url)
 data = resp.text
 print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
  gevent.spawn(f, 'https://www.python.org/'),
  gevent.spawn(f, 'https://www.yahoo.com/'),
  gevent.spawn(f, 'https://github.com/'),
])