zoukankan      html  css  js  c++  java
  • 线程的锁、信号量、事件、条件、定时器、线程队列

    同步锁

    import threading
    R=threading.Lock()
    R.acquire()
    '''
    对公共数据的操作
    '''
    R.release()
    from threading import Thread,Lock
    import os,time
    def work():
        global n
        lock.acquire()
        temp=n
        time.sleep(0.1)
        n=temp-1
        lock.release()
    if __name__ == '__main__':
        lock=Lock()
        n=100
        l=[]
        for i in range(100):
            p=Thread(target=work)
            l.append(p)
            p.start()
        for p in l:
            p.join()
        print(n) #结果肯定为0,由原来的并发执行变成串行,牺牲了执行效率保证了数据安全
    同步锁的引用
    #不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
    from threading import current_thread,Thread,Lock
    import os,time
    def task():
        #未加锁的代码并发运行
        time.sleep(3)
        print('%s start to run' %current_thread().getName())
        global n
        #加锁的代码串行运行
        lock.acquire()
        temp=n
        time.sleep(0.5)
        n=temp-1
        lock.release()
    
    if __name__ == '__main__':
        n=100
        lock=Lock()
        threads=[]
        start_time=time.time()
        for i in range(100):
            t=Thread(target=task)
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
        stop_time=time.time()
        print('主:%s n:%s' %(stop_time-start_time,n))
    
    '''
    Thread-1 is running
    Thread-2 is running
    ......
    Thread-100 is running
    主:53.294203758239746 n:0
    互斥锁与join

    死锁与递归锁

    进程也有死锁与递归锁,在进程那里忘记说了,放到这里一切说了额

    所谓死锁: 是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁

    from threading import Lock as Lock
    import time
    mutexA=Lock()
    mutexA.acquire()
    mutexA.acquire()
    print(123)
    mutexA.release()
    mutexA.release()
    死锁

    解决方法,递归锁,在Python中为了支持在同一线程中多次请求同一资源,python提供了可重入锁RLock。

    这个RLock内部维护着一个Lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次require。直到一个线程所有的acquire都被release,其他的线程才能获得资源。上面的例子如果使用RLock代替Lock,则不会发生死锁:

    from threading import RLock as Lock
    import time
    mutexA=Lock()
    mutexA.acquire()
    mutexA.acquire()
    print(123)
    mutexA.release()
    mutexA.release()
    递归锁

    典型问题:科学家吃面

    import time
    from threading import Thread,Lock
    noodle_lock = Lock()
    fork_lock = Lock()
    def eat1(name):
        noodle_lock.acquire()
        print('%s 抢到了面条'%name)
        fork_lock.acquire()
        print('%s 抢到了叉子'%name)
        print('%s 吃面'%name)
        fork_lock.release()
        noodle_lock.release()
    
    def eat2(name):
        fork_lock.acquire()
        print('%s 抢到了叉子' % name)
        time.sleep(1)
        noodle_lock.acquire()
        print('%s 抢到了面条' % name)
        print('%s 吃面' % name)
        noodle_lock.release()
        fork_lock.release()
    
    for name in ['哪吒','egon','yuan']:
        t1 = Thread(target=eat1,args=(name,))
        t2 = Thread(target=eat2,args=(name,))
        t1.start()
        t2.start()
    死锁问题
    import time
    from threading import Thread,RLock
    fork_lock = noodle_lock = RLock()
    def eat1(name):
        noodle_lock.acquire()
        print('%s 抢到了面条'%name)
        fork_lock.acquire()
        print('%s 抢到了叉子'%name)
        print('%s 吃面'%name)
        fork_lock.release()
        noodle_lock.release()
    
    def eat2(name):
        fork_lock.acquire()
        print('%s 抢到了叉子' % name)
        time.sleep(1)
        noodle_lock.acquire()
        print('%s 抢到了面条' % name)
        print('%s 吃面' % name)
        noodle_lock.release()
        fork_lock.release()
    
    for name in ['哪吒','egon','yuan']:
        t1 = Thread(target=eat1,args=(name,))
        t2 = Thread(target=eat2,args=(name,))
        t1.start()
        t2.start()
    递归锁解决死锁问题

    信号量

    同进程的一样

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

    实例:(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5):

    from threading import Thread,Semaphore
    import threading
    import time
    # def func():
    #     if sm.acquire():
    #         print (threading.currentThread().getName() + ' get semaphore')
    #         time.sleep(2)
    #         sm.release()
    def func():
        sm.acquire()
        print('%s get sm' %threading.current_thread().getName())
        time.sleep(3)
        sm.release()
    if __name__ == '__main__':
        sm=Semaphore(5)
        for i in range(23):
            t=Thread(target=func)
            t.start()
    实例
    与进程池是完全不同的概念,进程池Pool(4),最大只能产生4个进程,而且从头到尾都只是这四个进程,不会产生新的,而信号量是产生一堆线程/进程

    事件

    同进程的一样

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

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

    例如,有多个工作线程尝试链接MySQL,我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器,如果连接不成功,都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

    import threading
    import time,random
    from threading import Thread,Event
    
    def conn_mysql():
        count=1
        while not event.is_set():
            if count > 3:
                raise TimeoutError('链接超时')
            print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count))
            event.wait(0.5)
            count+=1
        print('<%s>链接成功' %threading.current_thread().getName())
    
    
    def check_mysql():
        print('33[45m[%s]正在检查mysql33[0m' % threading.current_thread().getName())
        time.sleep(random.randint(2,4))
        event.set()
    if __name__ == '__main__':
        event=Event()
        conn1=Thread(target=conn_mysql)
        conn2=Thread(target=conn_mysql)
        check=Thread(target=check_mysql)
    
        conn1.start()
        conn2.start()
        check.start()
    实例

    条件

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

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

    import threading
    
    def run(n):
        con.acquire()
        con.wait()
        print("run the thread: %s" % n)
        con.release()
    
    if __name__ == '__main__':
    
        con = threading.Condition()
        for i in range(10):
            t = threading.Thread(target=run, args=(i,))
            t.start()
    
        while True:
            inp = input('>>>')
            if inp == 'q':
                break
            con.acquire()
            con.notify(int(inp))
            con.release()
            print('****')
    实例

    定时器

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

    from threading import Timer
    def hello():
        print("hello, world")
    t = Timer(1, hello)
    t.start()  # after 1 seconds, "hello, world" will be printed

    线程对列

    queue队列 :使用import queue,用法与进程Queue一样

    queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

    class queue.Queue(maxsize=0) #先进先出

    import queue
    
    q=queue.Queue()
    q.put('first')
    q.put('second')
    q.put('third')
    
    print(q.get())
    print(q.get())
    print(q.get())
    '''
    结果(先进先出):
    first
    second
    third
    '''
    queue.Queue()

    class queue.LifoQueue(maxsize=0) #last in fisrt out

    import queue
    
    q=queue.LifoQueue()
    q.put('first')
    q.put('second')
    q.put('third')
    
    print(q.get())
    print(q.get())
    print(q.get())
    '''
    结果(后进先出):
    third
    second
    first
    '''
    后进先出

    class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列

    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')
    '''
    优先级队列
  • 相关阅读:
    BPM平台在企业业务系统中使用的价值讨论
    零售餐饮行业的信息化建设
    LINQ 与Oracle应用 :转帖
    k2之于.NET流程应用开发者
    利用xslt导出复杂样式的excel,支持多个worksheet
    利用偏移量快速定位数据内容
    简单天气项目中观察者模式解析
    作业3:基于墨刀的:视频剪辑软件原型设计
    必做作业2:视频剪辑软件调研
    .Net Core项目依赖项问题
  • 原文地址:https://www.cnblogs.com/caoshitong/p/9699585.html
Copyright © 2011-2022 走看看