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

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 -*-

"""
一个基于thread和queue的线程池，以任务为队列元素，动态创建线程，重复利用线程，
通过close和terminate方法关闭线程池。
"""
import queue
import threading
import contextlib
import time

# 创建空对象,用于停止线程
StopEvent = object()

def callback(status, result):
    """
    根据需要进行的回调函数，默认不执行。
    :param status: action函数的执行状态
    :param result: action函数的返回值
    :return:
    """
    pass

def action(thread_name,arg):
    """
    真实的任务定义在这个函数里
    :param thread_name: 执行该方法的线程名
    :param arg: 该函数需要的参数
    :return:
    """
    # 模拟该函数执行了0.1秒
    time.sleep(0.1)
    print("第%s个任务调用了线程 %s，并打印了这条信息！" % (arg+1, thread_name))

class ThreadPool:

    def __init__(self, max_num, max_task_num=None):
        """
        初始化线程池
        :param max_num: 线程池最大线程数量
        :param max_task_num: 任务队列长度
        """
        # 如果提供了最大任务数的参数，则将队列的最大元素个数设置为这个值。
        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 put(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):
        """
        创建一个线程
        """
        # 每个线程都执行call方法
        t = threading.Thread(target=self.call)
        t.start()

    def call(self):
        """
        循环去获取任务函数并执行任务函数。在正常情况下，每个线程都保存生存状态，
        直到获取线程终止的flag。
        """
        # 获取当前线程的名字
        current_thread = threading.currentThread().getName()
        # 将当前线程的名字加入已实例化的线程列表中
        self.generate_list.append(current_thread)
        # 从任务队列中获取一个任务
        event = self.q.get()
        # 让获取的任务不是终止线程的标识对象时
        while event != StopEvent:
            # 解析任务中封装的三个参数
            func, arguments, callback = event
            # 抓取异常，防止线程因为异常退出
            try:
                # 正常执行任务函数
                result = func(current_thread, *arguments)
                success = True
            except Exception as e:
                # 当任务执行过程中弹出异常
                result = None
                success = False
            # 如果有指定的回调函数
            if callback is not None:
                # 执行回调函数，并抓取异常
                try:
                    callback(success, result)
                except Exception as e:
                    pass
            # 当某个线程正常执行完一个任务时，先执行worker_state方法
            with self.worker_state(self.free_list, current_thread):
                # 如果强制关闭线程的flag开启，则传入一个StopEvent元素
                if self.terminal:
                    event = StopEvent
                # 否则获取一个正常的任务，并回调worker_state方法的yield语句
                else:
                    # 从这里开始又是一个正常的任务循环
                    event = self.q.get()
        else:
            # 一旦发现任务是个终止线程的标识元素，将线程从已创建线程列表中删除
            self.generate_list.remove(current_thread)

    def close(self):
        """
        执行完所有的任务后，让所有线程都停止的方法
        """
        # 设置flag
        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)

    # 该装饰器用于上下文管理
    @contextlib.contextmanager
    def worker_state(self, state_list, worker_thread):
        """
        用于记录空闲的线程，或从空闲列表中取出线程处理任务
        """
        # 将当前线程，添加到空闲线程列表中
        state_list.append(worker_thread)
        # 捕获异常
        try:
            # 在此等待
            yield
        finally:
            # 将线程从空闲列表中移除
            state_list.remove(worker_thread)

# 调用方式
if __name__ == '__main__':
    # 创建一个最多包含5个线程的线程池
    pool = ThreadPool(5)
    # 创建100个任务，让线程池进行处理
    for i in range(100):
        pool.put(action, (i,), callback)
    # 等待一定时间，让线程执行任务
    time.sleep(3)
    print("-" * 50)
    print("\033[32;0m任务停止之前线程池中有%s个线程，空闲的线程有%s个！\033[0m"
          % (len(pool.generate_list), len(pool.free_list)))
    # 正常关闭线程池
    pool.close()
    print("任务执行完毕，正常退出！")
    # 强制关闭线程池
    # pool.terminate()
    # print("强制停止任务！")
    

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/'),
])