码迷,mamicode.com
首页 > 编程语言 > 详细

并发编程——线程(二)

时间:2018-05-16 18:42:42      阅读:250      评论:0      收藏:0      [点我收藏+]

标签:2-2   参数   div   while   end   问题   []   join()   iss   

信号量:

  同进程的一样,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的用法

 

并发编程——线程(二)

标签:2-2   参数   div   while   end   问题   []   join()   iss   

原文地址:https://www.cnblogs.com/stfei/p/9047067.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!