python线程、协程

py3study

发布于 2020-01-13 10:42:21

2700

发布于 2020-01-13 10:42:21

文章被收录于专栏：python3

线程

Threading用于提供线程相关的操作，线程是应用程序中工作的最小单元。

更多方法：

start 线程准备就绪，等待CPU调度
setName 为线程设置名称
getName 获取线程名称
setDaemon 设置为后台线程或前台线程（默认）如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完后，等待前台线程也执行完成后，停止
join 逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义
run 线程被cpu调度后执行Thread类对象的run方法

直接调用

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time
  
def show(arg):
    time.sleep(1)
    print('thread'+str(arg))
t_list = []  
for i in range(10):
    t = threading.Thread(target=show, args=(i,))
    t.start()
    t_list.append(t) 
for i in t_list:
    t.join()        #等待所有线程结束结束程序
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 time
import threading

def run(n):

    print('[%s]------running----\n' % n)
    time.sleep(2)
    print('--done--')

def main():
    for i in range(5):
        t = threading.Thread(target=run,args=[i,])
        #time.sleep(1)
        t.start()
        t.join(1)
        print('starting thread', t.getName())
m = threading.Thread(target=main,args=[])
m.setDaemon(True) #将主线程设置为Daemon线程,它退出时,其它子线程会同时退出,不管是否执行完任务
m.start()
#m.join(timeout=2)
print("---main thread done----")

线程锁

如果同一时间有多个线程操作同一数据可能会出现混乱现象，所以需要加线程锁保证同一时间只有一个线程操作这一数据，保证数据一致性

加锁：lock = threading.RLock() 解锁：lock.release()

lock.acquire()

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

import threading
import time

gl_num = 0

lock = threading.Lock()

def Func():
    lock.acquire()   #修改数据前加锁
    global gl_num
    gl_num +=1      #对此公共变量进行-1操作
    time.sleep(1)
    print gl_num
    lock.release()  #修改后释放

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

正常来讲，这个num结果应该是0，但在python 2.7上多运行几次，会发现，最后打印出来的num结果不总是0，为什么每次运行的结果不一样呢？哈，很简单，假设你有A,B两个线程，此时都要对num 进行减1操作，由于2个线程是并发同时运行的，所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算，当A线程去处完的结果是99，但此时B线程运算完的结果也是99，两个线程同时CPU运算的结果再赋值给num变量后，结果就都是99。那怎么办呢？很简单，每个线程在要修改公共数据时，为了避免自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁，这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。

RLock（递归锁）

说白了就是在一个大锁中还要再包含子锁

import threading,time

def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num +=1
    lock.release()
    return num
def run2():
    print("grab the second part data")
    lock.acquire()
    global  num2
    num2+=1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res,res2)


if __name__ == '__main__':

    num,num2 = 0,0
    lock = threading.RLock()
    for i in range(10):
        t = threading.Thread(target=run3)
        t.start()

while threading.active_count() != 1:
    print('active num:',threading.active_count())
else:
    print('----all threads done---')
    print(num,num2)

Semaphore(信号量)

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

import threading,time

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

if __name__ == '__main__':

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

while threading.active_count() != 1:
    pass #print threading.active_count()
else:
    print('----all threads done---')
    print(num)

Events

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

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

clear：将“Flag”设置为False
set：将“Flag”设置为True

import threading,time
def light():
    if not event.isSet():
        event.set() #绿灯状态
    count = 0
    while True:
        if count < 10:
            print('\033[42;1m--green light on---\033[0m')
        elif count <13:
            print('\033[43;1m--yellow light on---\033[0m')
        elif count <20:
            if event.isSet():
                event.clear()
            print('\033[41;1m--red light on---\033[0m')
        else:
            count = 0
            event.set() #打开绿灯
        time.sleep(1)
        count +=1
def car(n):
    while 1:
        time.sleep(1)
        if  event.isSet(): #绿灯
            print("car [%s] is running.." % n)
        else:
            print("car [%s] is waiting for the red light.." %n)
            event.wait()
if __name__ == '__main__':
    event = threading.Event()
    Light = threading.Thread(target=light)
    Light.start()
    for i in range(3):
        t = threading.Thread(target=car,args=(i,))
        t.start()

queue队列

queue.qsize() 返回队列的大小 queue.empty() 如果队列为空，返回True,反之False queue.full() 如果队列满了，返回True,反之False queue.full 与 maxsize 大小对应 queue.get([block[, timeout]])获取队列，timeout等待时间 queue.get_nowait() 相当queue.get(False) queue.put(item, block=True, timeout=None) 非阻塞写入队列，timeout等待时间 queue.put_nowait(item) 相当queue.put(item, False) queue.task_done() 在完成一项工作之后，queue.task_done()函数向任务已经完成的队列发送一个信号 queue.join() 实际上意味着等到队列为空，再执行别的操作

class queue.Queue(maxsize=0) #先入先出
class queue.LifoQueue(maxsize=0) #last in fisrt out
class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列(priority_number, data)

1.多线程采用的是分时复用技术，即不存在真正的多线程，cpu做的事是快速地切换线程，以达到类似同步运行的目的，因为高密集运算方面多线程是没有用的，但是对于存在延迟的情况（延迟IO，网络等）多线程可以大大减少等待时间，避免不必要的浪费。

2.原子操作：这件事情是不可再分的，如变量的赋值，不可能一个线程在赋值，到一半切到另外一个线程工作去了……但是一些数据结构的操作，如栈的push什么的，并非是原子操作，比如要经过栈顶指针上移、赋值、计数器加1等等，在其中的任何一步中断，切换到另一线程再操作这个栈时，就会产生严重的问题，因此要使用锁来避免这样的情况。比如加锁后的push操作就可以认为是原子的了……

3.阻塞：所谓的阻塞，就是这个线程等待，一直到可以运行为止。最简单的例子就是一线程原子操作下，其它线程都是阻塞状态，这是微观的情况。对于宏观的情况，比如服务器等待用户连接，如果始终没有连接，那么这个线程就在阻塞状态。同理，最简单的input语句，在等待输入时也是阻塞状态。

4.在创建线程后，执行p.start()，这个函数是非阻塞的，即主线程会继续执行以后的指令，相当于主线程和子线程都并行地执行。所以非阻塞的函数立刻返回值的～

对于资源，加锁是个重要的环节。因为python原生的list,dict等，都是not thread safe的。而Queue，是线程安全的，因此在满足使用条件下，建议使用队列。

队列适用于 “生产者-消费者”模型。双方无论数量多少，产生速度有何差异，都可以使用queue。

import time,queue,threading
def consumer(n):
    while True:
        print('consumer [%s] get task: [%s]' % (n,q.get()))
        time.sleep(1)
        q.task_done()
def producer(n):
    count = 1
    while True:
        print('producer [%s] produced a new task :%s ' % (n,count))
        q.put(count)
        count +=1
        q.join()
        print('task finished')
q = queue.Queue()
c1 = threading.Thread(target=consumer,args=[1,])
c2 = threading.Thread(target=consumer,args=[2,])
c3 = threading.Thread(target=consumer,args=[3,])

p1 = threading.Thread(target=producer,args=['user1',])
p2 = threading.Thread(target=producer,args=['user2',])
p3 = threading.Thread(target=producer,args=['user3',])

c1.start()
p1.start()

在上面的例子中，Producer在随机的时间内生产一个“产品”，放入队列中。Consumer发现队列中有了“产品”，就去消费它。本例中，由于Producer生产的速度快于Consumer消费的速度，所以往往Producer生产好几个“产品”后，Consumer才消费一个产品。

Queue模块实现了一个支持多producer和多consumer的FIFO队列。当共享信息需要安全的在多线程之间交换时，Queue非常有用。Queue的默认长度是无限的，但是可以设置其构造函数的maxsize参数来设定其长度。Queue的put方法在队尾插入，该方法的原型是：

put( item[, block[, timeout]])

如果可选参数block为true并且timeout为None（缺省值），线程被block，直到队列空出一个数据单元。如果timeout大于0，在timeout的时间内，仍然没有可用的数据单元，Full exception被抛出。反之，如果block参数为false（忽略timeout参数），item被立即加入到空闲数据单元中，如果没有空闲数据单元，Full exception被抛出。

Queue的get方法是从队首取数据，其参数和put方法一样。如果block参数为true且timeout为None（缺省值），线程被block，直到队列中有数据。如果timeout大于0，在timeout时间内，仍然没有可取数据，Empty exception被抛出。反之，如果block参数为false（忽略timeout参数），队列中的数据被立即取出。如果此时没有可取数据，Empty exception也会被抛出。

协程

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

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

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：

协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。

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

协程的好处：

无需线程上下文切换的开销
无需原子操作锁定及同步的开销
方便切换控制流，简化编程模型
高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

缺点：

无法利用多核资源：协程的本质是个单线程,它不能同时将单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
进行阻塞（Blocking）操作（如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

gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。

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

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2019/08/11 ，如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

编程算法

登录后参与评论

0 条评论

热度