并发编程——线程

Posted 2020-11-05 Qingqiu_Gu

信号量：

　　同进程的一样，Semaphore管理一个内置的计数器，每当调用acquire()时内置计数器-1，调用release()时内置计数器+1；计数器不能小于0；当计数器为0，acquire()将阻塞线程直到其他线程调用release()。

import time
from threading import Thread,Semaphore

def func(sema,i):
    sema.acquire()
    print(i)
    time.sleep(1)
    sema.release()

sema = Semaphore(5)
for i in range(20):
    Thread(target=func,args=(sema,i)).start()

信号量例子

事件：

同进程的一样

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。

import time
import random
from threading import Event,Thread

# 连接数据库
def connect_db(e):
    count = 1
    while count <= 3:
        print(‘尝试第%s次检测连接‘%count)
        e.wait(0.5)
        # 如果不传参会一直等到事件为True为止
        # 如果传参数，传一个时间参数
        count += 1
        if e.is_set():
            print(‘连接成功‘)
            break
    else:
        print(‘连接失败‘)

def check_conn(e):
    ‘‘‘检测数据库是否可以连接‘‘‘
    time.sleep(random.randint(1,2))
    e.set()

e = Event()
Thread(target=check_conn,args=(e,)).start()
Thread(target=connect_db,args=(e,)).start()

实例

条件：

　　使得线程等待，只有满足条件时，才释放n个线程。

Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。

详细说明

from threading import Condition,Thread

def func(i,con):
    con.acquire()
    con.wait()
    print(i*‘*‘)
    con.release()

con = Condition()
for i in range(10):
    Thread(target=func,args=(i,con)).start()
while True:
    n = int(input(‘>>>‘))
    con.acquire()
    con.notify(n)   
    con.release()
    
# 内部条件控制wait的行为，n为数字，可以按n次控制线程的走向。
# n就是wait将释放几次。

例子

定时器：

　　定时器，指定n秒后执行某个操作。

from threading import Timer

def func():
    print(‘*‘*10)

t = Timer(5,func)   # 要开启一个线程，等到5秒后才开启并且执行。
t.start()

线程队列：

　　queue队列：使用import queue，用法与进程Queue一样。

import queue

q = queue.Queue()   # 默认先进先出
q.put(1)
q.put(2)
q.put(3)
print(q.get())  # 1
print(q.get())  # 2
print(q.get())  # 3

先进先出

import queue

q = queue.LifoQueue()
q.put(1)
q.put(2)
q.put(3)

print(q.get())  # 3
print(q.get())  # 2
print(q.get())  # 1

后进先出

　　存储数据时可设置优先级的队列

import queue

q = queue.PriorityQueue()
# put进入一个元祖，元组的第一个元素时优先级(通常是数字，也可以是非数字之间的比较)数字越小优先级越高。
q.put((20,‘a‘))
q.put((10,‘b‘))
q.put((30,‘c‘))

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

‘‘‘
(10,‘b‘)
(20,‘a‘)
(30,‘c‘)
‘‘‘
#  数字越小优先级越高,优先级高的优先出队

优先级队列

python标准模块——concurrent.futures

#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数

import os
import time
import random
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

def func(i):
    print(i*‘*‘)
    time.sleep(1)
    return i**2

def call_back(arg):
    print(arg.result()*‘-‘)

if __name__ == ‘__main__‘:
    thread_pool = ThreadPoolExecutor(5)
    # ret_l = []
    for i in range(10):
        thread_pool.submit(func,i).add_done_callback(call_back)
        # ret = thread_pool.submit(func,i).add_done_callback(call_back)
        # ret_l.append(ret)
    # thread_pool.shutdown() # 默认是wait = True 等待线程执行完毕才会继续往下执行代码# 相当于# close+join
    thread_pool.shutdown(wait=False)     # 不等待线程全部执行完毕
    # for ret in ret_l:
    #     print(ret.result())
    print(‘wahaha‘)

例子

import os
import time
import random
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

def task(n):
    print(‘%s is runing‘ % os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == ‘__main__‘:
    executor = ThreadPoolExecutor(3)
    # for i in range(11):
    #     future = executor.submit(task,i)
    executor.map(task,range(1,12))  # 相当于上面两行代码。

map的用法