文成小盆友python-num11－（1） 线程 进程 协程

本节主要内容

• 线程补充
• 进程
• 协程

一.线程补充

1.两种使用方法

这里主要涉及两种使用方法，一种为直接使用，一种为定义自己的类然后继承使用如下：

• 直接使用如下：

import  threading

def f1(arg):
    print(arg)

t = threading.Thread(target=f1,args=(123,))    ＃直接使用
t.start()

• 自定义类方法：

#2.自定义方法。

class Mythread(threading.Thread):  ＃继承threading.Thread类
    def __init__(self,func,args):        ＃自己的构造函数
        self.func = func
        self.args = args
        super(Mythread,self).__init__() ＃执行父的构造函数（如不需要刻意不执行）

    def run(self):                            ＃内部会先调用这个方法。此处可以自定义
        self.func(self.args)


obj = Mythread(f1,123)                 ＃创建对象。
obj.start()    
        

2.python自带队列应用

queue 模块在python中提供队列的应用其中队列的种类有4种如下：

•      先去先出队列  queue.Queue(5)
•      后进先出队列 queue.LifoQueue()
•      优先级队列  queue.PriorityQueue()
•      双向队列  queue.deque()

以先进先出队列为例子，简单说明参数的使用：

##先进先出。
import queue

q = queue.Queue(5)  #最大放5个数据

q.put(11)                  ＃put：直接往队列中放数据
q.put(22)
print(q.full())    #full判读是否为满，不满足最大的5个值 - 所以不满,输出值为False。
print(q.empty())   #empty判断是否为空，此处不空 输出值为FALSE
print(q.get())    ＃get 在队列中获取数据（按照队列的顺序）
print(q.get())

print(q.empty()) #此处空 输出的值为TRUE
print(q.full())

q.put(33,block=False)    ＃block 是否阻塞
q.put(33,block=True,timeout=2) ＃time 等待时间。

print(q.get())             
print(q.get())
#print(q.get(block=False,))  ###--- 如过没有此值则会在这里阻塞(因为已经取空了)
    

如上为队列使用的主要方法。其它类型队列使用相同，如下：

##后进先出队列
import queue
q = queue.LifoQueue()
q.put(123)
q.put(456)

print(q.get())    ##输出456

##优先级队列

import  queue
q = queue.PriorityQueue()

q.put((1,111111))
q.put((0,222222))    #前面值越小优先级越低
q.put((5,333333))
q.put((9,444444))

print(q.get())

# ##双向队列
q = queue.deque()
q.append(123)
q.append(456)
q.appendleft(111111)

print(q.pop())
print(q.popleft())    # 输出最左边的 111111

3.实例练习－利用多线程＋队列实现一个生产者，消费者模型。

＃＃＃
import queue
import threading
import time

q = queue.Queue(20)     #先创建一个队列

def producter(arg):     #创建一个函数-生产者-(将任务,提交到队列中供消费者获取)
    """
    买票提交订单到队列中
    :param arg:
    :return:
    """
    while True:
        q.put(str(arg) + '--票')    #用队列的put方法将任务放到,队列中


for i in range(300):          #模拟300个客户端来提交任务(比如买票)
    t = threading.Thread(target=producter,args=(i,))    #每个客户端创建一个线程去提交任务
    t.start()

def consumer(arg):                  #创建一个函数 - 消费者 -(在队列中获取任务然后处理)
    """
    在队列中获取订单信息然后处理
    :param arg:
    :return:
    """
    while True:
        print(arg,q.get())         #不停的在队列中获取任务。
        time.sleep(2)               #模拟此程序的处理的时间为两秒

for j in range(3):                 #模拟有3个消费者
    t = threading.Thread(target=consumer,args=(j,))
    t.start()

4.线程中的锁机制

• 线程锁机制的出现场景：

由于线程之间是随机调度的，并且当多个线程同时修改一条数据的时候，可能会出现脏数据所以出现了线程锁，同一时刻只允许某几个线程进程操作。

下面看一个例子没有使用锁的情况下：

#无锁的情况
import time,threading

NUM = 10         #设定一个全局变量 - 供多个线程修改

def func(): #设置一个函数,来修改全局变量
    global NUM
    NUM -= 1
    time.sleep(1)   ＃模拟处理用时需要1秒

    print(NUM)

for i in range(10):  ＃创建多个线程，同时修改全局变量
    t = threading.Thread(target=func,)
    t.start()

＃＃＃＃显示如下：
0
0
0
0
0
0
0
0
0
0

Process finished with exit code 0

分析： 此程序的目的是每次让全局变量自减去1然后打印出来，很明显输出结果并不是我们想要的。原因是当一个线程减去值后还没有输出其它的线程也获得了这个值然后又在这个基础上做了操作，。。。最后当所有线程都减掉1后值为0   然后大家一起输出最后所有的输出值都是0。下面引入线程的锁机制就可以限制这个情况。

Lock、RLock   － 线程锁机制

下面上一段代码：采用锁机制

• 互斥锁

#互斥锁 － 同时一个线程运行
import time,threading

NUM = 10         #设定一个全局变量 - 供多个线程修改

def func(lock): #设置一个函数,来修改全局变量
    # #上锁    上锁后其它线程只能等待解锁后才能继续处理
    lock.acquire()   
    global NUM
    NUM -= 1
    time.sleep(1)
    # #解锁
    lock.release()

    print(NUM)

lock = threading.RLock()       #
# lock = threading.BoundedSemaphore(5)  #信号量  一次放过5个

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

＃＃＃＃显示如下
9
8
7
6
5
4
3
2
1
0

Process finished with exit code 0

如上的代码输出的结果是我们最终想要的。依次输出了数字递减后的数字。如上的锁的方式为互斥锁，同时只能允许一个线程在工作。下面看下其它的类型：

• 信号量

Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许5个人上厕所，后面的人只能等里面有人出来了才能再进去。

#  信号量
import time,threading

NUM = 10         #设定一个全局变量 - 供多个线程修改

def func(lock): #设置一个函数,来修改全局变量
    # #上锁
    lock.acquire()
    global NUM
    NUM -= 1
    time.sleep(1)
    # #解锁
    lock.release()

    print(NUM)

lock = threading.BoundedSemaphore(5)  #信号量  一次放过5个

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

＃＃＃如上代码一次允许5个线程同时工作，等这5个都出来后然后后面的进程才能够进入处理。

• 事件锁（event）

python中的事件主要用户主线程来控制其它子线程，事件主要提供了3个方法：

1. 1. wait()    执行时判断是否等待
   2. set()　　将 flag 设置为true
   3. clear()   将 flag 设置为false

event处理机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

如下代码实例：

#event判断（全部放或者不放）:

import  threading

def func(i,e):     ＃创建一个函数让线程执行
    print(i)

    e.wait()   #等待 运行到此处后直接等待，然后判断flag是否为false －默认为false
    print(i+100)   ＃等待后的执行命令

envent = threading.Event()

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

    t.start()

envent.clear() #设置成红灯  -  默认是红灯。 （clear就相当于红灯）
inp = input(">>>")

if inp == "1":
    envent.set()   #设置成绿灯 如果是set（）则就执行等待处后面的语句
＃＃＃＃＃＃显示结果如下：
0
1
2
3
4
5
6
7
8
9                ＃执行到此处时等待
>>>1         ＃如果输入为1 这 e.set() 设置为绿灯。再执行后面的内容
100
104
106
107
103
105
109
101
102
108

Process finished with exit code 0

• 条件锁（Condition）

使得线程等待，只有当满足某个条件时才放行某几个线程任务：

第一种方式：

# 条件的方式判断是否应该放行
#1.第一种方式:  con.wait()
import threading      #倒入线程模块
def func(i,con):      #创建一个函数 - 被线程执行
    print(i)
    con.acquire()     #上锁,
    con.wait()        #使线程等待
    print(i+100)      #等待条件成立了后的输出
    con.release()     #解锁

c = threading.Condition()

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

while True:
    inp = input(">>>")
    if inp == "q":  
        break
    c.acquire()            ＃此时已经有10个线程在排队等待放行
    c.notify(int(inp))     #唤醒指定数量再等待的线程,默认参数为1个 在执行前必须是已经被锁了,否则会抛出错误
    c.release()

第二种方式：

wait_for()

#第二种方式:
import threading
def condition():   #创建一个函数用于返回值。
    ret = False    
    r = input(">>")
    if r == 'true':   #判断是否为true
        ret = True
    else: 
        ret = False
    return ret

def func(i,con):
    print(i)
    con.acquire()    #先上锁 
    con.wait_for(condition)  #等待结果 如果为true则继续执行
    print(i+100)
    con.release()

c = threading.Condition()

for i in range(10):
    t = threading.Thread(target=func,args=(i,c,))
    t.start()
＃＃＃＃显示结果如下：
0
>>1
2
3
4
5
6
7
8
9
true
100
>>

以上为线程中的锁机制。

5.附加Timer

线程中的定时器，指定n秒后去执行某个操作。就如下，等待两秒后去执行制定的函数。

# ##### timer
from threading import Timer


def hello():
    print("hello, world")

t = Timer(2, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

6.简单实现线程池。

python中没有提供直接可用的线程池，所以要在python中使用线程池 1.使用第三放模块，2.自己动手丰衣足食。 下面是一个简单的线程池实现。

#########实现简单线程池

import queue
import threading
import time

class ThreadPool:                    #创建一个类
    def __init__(self,maxsize=5):
        '''
        构造方法
        :param maxsize: 队列中最多成员数量。
        '''
        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)

pool = ThreadPool(5)                      #创建对象

def task(arg,p):                          #创建一个函数让线程来执行
    print(arg)
    time.sleep(1)
    p.add_thread()                        #每个线程工作完后都在把一个类归还到队列中

for i in range(100):

    t = pool.get_thread()                 #每次拿一个(当队列中拿完了就等待,其它的归还后在拿然后在执行)
    obj = t(target=task,args=(i,pool,))
    obj.start()

以上为线程部分补充。

！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！

二.进程

1.创建多进程

创建多个进程如下方式（跟创建线程的方法类似，只是库名不同，其它使用方法也是类似的）

进程与进程之间默认数据是不共享的，各自持用一份，所以创建进程的开销是很大的。

from multiprocessing import Process  ＃倒入模块
import time


def foo(i):
    print('say hi', i)


for i in range(10):
    p = Process(target=foo, args=(i,))   ＃创建方式同线程的创建方式
    p.start()

＃＃显示如下：
say hi 0
say hi 1
say hi 2
say hi 3
say hi 4
say hi 5
say hi 6
say hi 7
say hi 8
say hi 9

Process finished with exit code 0

2.实现进程间数据的共享

进程之间各自持有一份数据，默认是不共享的。我门来看一下如下的例子：

from multiprocessing import Process

li = []


def foo(i):
    li.append(i)
    print('say hi', li)


for i in range(10):
    p = Process(target=foo, args=(i,))
    p.start()

print('ending', li)

＃＃＃＃＃
say hi [0]
say hi [1]
say hi [2]
say hi [3]
say hi [4]
say hi [5]
say hi [6]
say hi [7]
ending []
say hi [8]
say hi [9]

Process finished with exit code 0

＃由上面的显示可以看出来，各自维护着各自的数据，列表并没有累加

使用如下方法可以实现进程之间的数据共享：

• 1.queues  

#第一种共享方法 queues
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing

def foo(i,arg):
    arg.put(i)
    #arg.append(1)
    #print('say hi',i,li)
    print('say hi',i,arg)

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

＃＃＃＃＃显示如下，arg的内存地址为相同的，可见操作了相同的数据

say hi 0 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 1 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 2 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 3 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 4 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 5 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 6 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 7 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 8 <multiprocessing.queues.Queue object at 0x101377ef0>
say hi 9 <multiprocessing.queues.Queue object at 0x101377ef0>

Process finished with exit code 0

• 2.Array

from multiprocessing import Process
from multiprocessing import Array

def foo(i,arg):
    arg[i] = i + 100
    for item in arg:
        print(item)
    print("__++++__")

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

显示如下：

100
0
0
0
0
0
0
0
0
0
__++++__
100
101
0
0
0
0
0
0
0
0
__++++__
100
101
102
0
0
0
0
0
0
0
__++++__
100
101
102
103
0
0
0
0
0
0
__++++__
100
101
102
103
104
0
0
0
0
0
__++++__
100
101
102
103
104
105
0
0
0
0
__++++__
100
101
102
103
104
105
106
0
0
0
__++++__
100
101
102
103
104
105
106
107
0
0
__++++__
100
101
102
103
104
105
106
107
108
0
__++++__
100
101
102
103
104
105
106
107
108
109
__++++__

Process finished with exit code 0

＃＃＃可以看到所有的进程操作的都是一个数据

View Code

Array 中数据类型对应表：

'c': ctypes.c_char,  'u': ctypes.c_wchar,
    'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
    'h': ctypes.c_short, 'H': ctypes.c_ushort,
    'i': ctypes.c_int,   'I': ctypes.c_uint,
    'l': ctypes.c_long,  'L': ctypes.c_ulong,
    'f': ctypes.c_float, 'd': ctypes.c_double

类型对应表

• 3.manage.dict（）

from multiprocessing import Manager
from multiprocessing import Process

def foo(i,arg):
    arg[i] = i + 100
    print(arg.values())

if __name__ == "__main__":
    obj = Manager()
    li = obj.dict()

    for i in range(10):
        p = Process(target=foo,args=(i,li))
        p.start()
        p.join()

＃＃显示如下：
[100]
[100, 101]
[100, 101, 102]
[100, 101, 102, 103]
[100, 101, 102, 103, 104]
[100, 101, 102, 103, 104, 105]
[100, 101, 102, 103, 104, 105, 106]
[100, 101, 102, 103, 104, 105, 106, 107]
[100, 101, 102, 103, 104, 105, 106, 107, 108]
[100, 101, 102, 103, 104, 105, 106, 107, 108, 109]

Process finished with exit code 0
＃＃＃可见每次都是针对同一个字典进行操作的

多进程数据共享时又会在此同多线程一样会涉及到锁的问题如下一个实例：

当创建进程时（非使用时），共享数据会被拿到子进程中，当进程中执行完毕后，再赋值给原值。

from multiprocessing import Process, Array, RLock

def Foo(lock,temp,i):
    """
    将第0个数加100
    """
    lock.acquire()   #上锁
    temp[0] = 100+i
    for item in temp:
        print (i,'----->',item)
    lock.release()  #解锁

lock = RLock()
temp = Array('i', [11, 22, 33, 44])

for i in range(20):
    p = Process(target=Foo,args=(lock,temp,i,))
    p.start()

锁相关

3.进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池中有两个方法：

#####################进程池####################################
from multiprocessing import Pool
import time

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

if __name__ == '__main__':
    pool = Pool(5)

    for i in range(30):
        pool.apply_async(func=f1,args=(i,))

    pool.close()  #所有的任务执行完毕
    pool.join()

    #time.sleep(1)
    #pool.terminate()   #立即终止

三.协程

协程，又称微线程，纤程。英文名Coroutine。

协程的概念很早就提出来了，但直到最近几年才在某些语言（如Lua）中得到广泛应用。

子程序，或者称为函数，在所有语言中都是层级调用，比如A调用B，B在执行过程中又调用了C，C执行完毕返回，B执行完毕返回，最后是A执行完毕。

所以子程序调用是通过栈实现的，一个线程就是执行一个子程序。

子程序调用总是一个入口，一次返回，调用顺序是明确的。而协程的调用和子程序不同。

协程看上去也是子程序，但执行过程中，在子程序内部可中断，然后转而执行别的子程序，在适当的时候再返回来接着执行。

注意，在一个子程序中中断，去执行其他子程序，不是函数调用，有点类似CPU的中断。比如子程序A、B：

def A():
    print '1'
    print '2'
    print '3'

def B():
    print 'x'
    print 'y'
    print 'z'

假设由协程执行，在执行A的过程中，可以随时中断，去执行B，B也可能在执行过程中中断再去执行A，结果可能是如下：但是在A中是没有调用B的，这就是协程

1
2
x
y
3
z

协程的优势：优势就是不需要多线程的锁机制，因为只有一个线程，也不存在同时写变量冲突，在协程中控制共享资源不加锁，只需要判断状态就好了，所以执行效率比多线程高很多。

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

greenlet

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