铁乐学python_Day42_线程锁和队列

我会带着你远行 2022-05-23 20:25 141阅读 0赞

铁乐学python\_Day42\_锁和队列

例：多个线程抢占资源的情况
    from threading import Thread
    import time
    
    def work():
        global n
        temp = n
        time.sleep(0.1)
        n = temp - 1
    
    if __name__ == '__main__':
        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) # 很有可能n=99
    
    这个时候为了保障数据的安全，我们可以对公共数据使用锁锁起来。
    import threading
    R=threading.Lock()
    R.acquire()
    ''' 对公共数据的操作 '''
    R.release()
    
    例：加锁同步数据
    from threading import Thread,Lock
    import 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，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全。
    
    例：互斥锁与join的区别
    #不加锁:并发执行,速度快,数据不安全
    from threading import current_thread,Thread,Lock
    import time
    def task():
        global n
        print('%s is running' %current_thread().getName())
        temp=n
        time.sleep(0.5)
        n=temp-1
    
    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 主:0.5216062068939209 n:99 '''
    
    #部分数据加上同步锁:未加锁部分并发执行，加锁部分串行执行,速度慢,数据安全。
    from threading import current_thread,Thread,Lock
    import 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 '''
    
    #加锁会让运行变成串行,而在start之后立即使用join,不用加锁了也是串行的效果。那么为什么使用加锁呢？
    #在start之后立刻使用jion,也会将100个任务的执行变成串行,最终n的结果是0,也是安全的,
    #但问题是start后立即join:任务内的所有代码都是串行执行的,
    #而加锁,只是加锁的部分即修改共享数据的部分是串行的,
    #单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
    
    from threading import current_thread,Thread,Lock
    import time
    def task():
        time.sleep(3)
        print('%s start to run' %current_thread().getName())
        global n
        temp=n
        time.sleep(0.5)
        n=temp-1
    
    if __name__ == '__main__':
        n=100
        lock=Lock()
        start_time=time.time()
        for i in range(100):
            t=Thread(target=task)
            t.start()
            t.join()
        stop_time=time.time()
        print('主:%s n:%s' %(stop_time-start_time,n))
    
    ''' Thread-1 start to run Thread-2 start to run ...... Thread-100 start to run 主:350.6937336921692 n:0 #耗时是多么的恐怖 '''

## 死锁与递归锁死锁与递归锁 ##

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

如下就是死锁:
    from threading import Lock as Lock
    import time
    mutexA=Lock()
    mutexA.acquire()
    mutexA.acquire()
    print("铁乐猫")
    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("铁乐猫")
    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 ['tiele','mao','铁乐猫']:
        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 ['tiele','mao','铁乐猫]: t1 = Thread(target=eat1,args=(name,)) t2 = Thread(target=eat2,args=(name,)) t1.start() t2.start() 定时器(Timer) 定时器，指定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队列） ##

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

队列长度可为无限或者有限。可通过Queue的构造函数的可选参数maxsize来设定队列长度。  
如果maxsize小于1就表示队列长度无限。

调用队列对象的put()方法在队尾插入一个项目。  
put()有两个参数，第一个item为必需的，为插入项目的值；  
第二个block为可选参数，默认为1。  
如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。  
如果block为0，put方法将引发Full异常。

调用队列对象的get()方法从队头删除并返回一个项目。  
可选参数为block，默认为True。  
如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。  
如果队列为空且block为False，队列将引发Empty异常。

### python queue 模块有三种队列： ###

1）class queue.Queue(maxsize=0) #First In First Out ［FIFO 类似于栈 先进先出］ 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
    ''' 2）class queue.LifoQueue(maxsize=0) #Last In Fisrt Out ［LIFO 类似于堆 后进先出］ 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')
    ''' 队列中的其它方法： Queue.full # 与maxsize大小对应 Queue.qsize() # 返回队列的大小 Queue.empty() # 队列若为空则返回True Queue.full() # 队列若为满则返回True Queue.join() # block（阻塞）直到queue被消费完毕，再执行后面的操作 Queue.get([block[, timeout]]) # 获取队列，timeout等待时间 Queue.get_nowait() # 相当Queue.get(False) Queue.put(item) # 非阻塞 写入队列，timeout等待时间 Queue.put_nowait(item) # 相当Queue.put(item, False) Queue.task_done() # 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号 例：使用多线程完成队列中的任务 import queue import threading import time import random q = queue.Queue(0) # 当有多个线程共享一个东西的时候就可以用它了 NUM_WORKERS = 3 class MyThread(threading.Thread): def __init__(self, input, worktype): self._jobq = input self._work_type = worktype threading.Thread.__init__(self) def run(self): while True: if self._jobq.qsize() > 0: self._process_job(self._jobq.get(), self._work_type) else: break def _process_job(self, job, worktype): doJob(job, worktype) def doJob(job, worktype): time.sleep(random.random() * 3) print("doing", job, " worktype ", worktype) if __name__ == '__main__': print("begin....") for i in range(NUM_WORKERS * 2): q.put(i) # 放入到任务队列中去 print("job qsize:", q.qsize()) for x in range(NUM_WORKERS): MyThread(q, x).start() 运行效果： begin.... job qsize: 6 doing 2 worktype 2 doing 0 worktype 0 doing 1 worktype 1 doing 3 worktype 2 doing 5 worktype 1 doing 4 worktype 0

end  
参考：[http://www.cnblogs.com/Eva-J/articles/8306047.html][http_www.cnblogs.com_Eva-J_articles_8306047.html]

[http_www.cnblogs.com_Eva-J_articles_8306047.html]: http://www.cnblogs.com/Eva-J/articles/8306047.html