线程和进程

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
import time
import threading

def f0(a1,a2):
    print("f1")
    a = a1 +a2
    print(a)


def f1(a1,a2):
    time.sleep(5)
    f0(a1,a2)

t = threading.Thread(target= f1,args= (123,456))
t.setDaemon(True)
t.start()
t = threading.Thread(target= f1,args= (123,456))
t.start()
t = threading.Thread(target= f1,args= (123,456))
t.start()


"""

# python线程
"""
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
import time


class MyThread(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数

        print("running on number:%s" % self.num)

        time.sleep(3)


if __name__ == '__main__':
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

"""
# 线程的阻塞和释放
"""
import threading
import time
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 multiprocessing
import time

def f1(a1):
    time.sleep(2)
    print(a1)
if __name__ == "__main__":

    t = multiprocessing.Process(target= f1,args= (11,))
    t.start()

    t = multiprocessing.Process(target= f1,args= (11,))
    t.start()
    print('end')

# 2、daemon
# 若果 False
# True

"""

# from multiprocessing import Process
# from multiprocessing import Manager
#
# import time
#
# li = []
#
#
# def foo(i):
#     li.append(i)
#     print('say hi', li)
#
# if __name__ == '__main__':
#     for i in range(10):
#         p = Process(target=foo, args=(i,))
#         p.start()
#
#     print('ending', li)


"""
from  threading import Thread
import time
li = []
def f1(i):
    li.append(i)
    print("你好",li)
for i in range(10):
    t = Thread(target= f1,args=(i,))
    t.start()
print("ending",li)
"""

# 实现线程和进程的互通
"""
from multiprocessing import Process,Array


temp = Array('i',[11,22,33,44])#Array是一个数组

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

if __name__ == "__main__":
    for i in range(2):
        p = Process(target = Foo,args =(i,))
        p.start()

"""

# Manager进程和进程之间的参合
"""
import time
from multiprocessing import Process,Manager
def Foo(i,dic):
    dic[i] = 100+i
    print(len(dic))

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

"""

1、Threading 用于提供线程相关的操作，线城是应用程序中工作的最小单元，代码如下

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

2、自定义线程类

import threading
import time


class MyThread(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数

        print("running on number:%s" % self.num)

        time.sleep(3)


if __name__ == '__main__':
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

3、线程锁

# 未使用锁时如下
"""
import threading
import time

gl_num = 0
def show(a):
    global gl_num
    time.sleep(1)
    gl_num += 1
    print(gl_num)

if __name__ == "__main__":
    for i in range(10):
        t = threading.Thread(target= show,args= (i,))
        t.start()

print("main thread stop")
"""
# 用锁后
"""
import threading
import time
gl_num = 0

lock = threading.RLock()
def Func():
    lock.acquire()
    global gl_num
    gl_num +=1
    time.sleep(0.5)
    print(gl_num)
    lock.release()

for i in range(10):
    t = threading.Thread(target=Func,)
    t.start()

4、事件（event）

python 线程的时间主要用于主线程控制其他线程的执行，
事件主要提供了三个方法 set，wait，clear

事件处理的机制：全局定义了一个'Flag',如果'Flag'的值为False
，那么当程序执行event。wait方法时就会阻塞，如果Flag的值为True ，那么遇到even.wait
时便不会阻塞
clear:将'Flag'设置为False
set ：将'Flag'设置为True

import threading
def do(event):
    print("阻塞前")
    event.wait()
    print("阻塞后")

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

event_obj.clear()
inp = input("input")
if inp == True:
    event_obj.set()

5、进程池

from multiprocessing import Pool
import time
def f1(w):
    time.sleep(1)
    return 100

def f2(arg):
    print(arg)

if __name__ == "__main__":
    pool = Pool(5)
    for i in range(10):
        pool.apply_async(func= f1,args=(i,),callback= f2)
        print("主线程运行")

    pool.close()
    pool.join()

6、一个简单的线程池

import queue
import  threading
import time

class ThreadPool(object):

    def __init__(self,max_num):
        self.queue = queue.Queue(max_num)#创建一个队列
        for i in range(max_num):
            self.queue.put(threading.Thread)

    def get_thread(self):
        return self.queue.get()

    def add_thread(self):
        self.queue.put(threading.Thread)

def func(pool,a1):
    time.sleep(1)
    print(a1)
    pool.add_thread()
    

p = ThreadPool(10)
for i in range(100):
    thread = p.get_thread()
    t = thread(target = func,args = (p,i))#创建了一个线程

    t.start()

7、复杂的线程池

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.queue.clear()

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

8、协程

线程和进程的操作是由程序触发系统接口，最后的执行者是系统；协程的操作则是程序员。

协程存在的意义：对于多线程应用，CPU通过切片的方式来切换线程间的执行，线程切换时需要耗时（保存状态，下次继续）。协程，则只使用一个线程，在一个线程中规定某个代码块执行顺序。

协程的适用场景：当程序中存在大量不需要CPU的操作时（IO），适用于协程；

greenlet

#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
 
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()

gevent

import gevent

 

def foo():

    print('Running in foo')

    gevent.sleep(0)

    print('Explicit context switch to foo again')

 

def bar():

    print('Explicit context to bar')

    gevent.sleep(0)

    print('Implicit context switch back to bar')

 

gevent.joinall([

    gevent.spawn(foo),

    gevent.spawn(bar),

])

 

 

遇到IO操作自动切换

 

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

def f(url):
    print('GET: %s' % url)
    resp = urllib2.urlopen(url)
    data = resp.read()
    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/'),
])