Python 线程同步变量，同步条件，列队

条件变量同步

有一类线程需要满足条件之后才能够继续执行，Python提供了threading.Condition 对象用于条件变量线程的支持，它除了能提供RLock()或Lock()的方法外，还提供了 wait()、notify()、notifyAll()方法。 lock_con=threading.Condition([Lock/Rlock])：锁是可选选项，不传人锁，对象自动创建一个RLock()。

wait()：条件不满足时调用，线程会释放锁并进入等待阻塞；
notify()：条件创造后调用，通知等待池激活一个线程；
notifyAll()：条件创造后调用，通知等待池激活所有线程。


import threading, time
from random import randint

class Producer(threading.Thread):
    def run(self):
        global L
        while True:
            val = randint(0, 100)
            print('生产者', self.name, ':Append'+str(val),L)
            if lock_con.acquire():
                L.append(val)
                lock_con.notify()
                lock_con.release()
            time.sleep(3)

class Consumer(threading.Thread):
    def run(self):
        global L
        while True:
            lock_con.acquire()
            if len(L) == 0:
                lock_con.wait()
            print('消费者', self.name, ":Delete" + str(L[0]), L)
            del L[0]
            lock_con.release()
            time.sleep(0.25)


if __name__ == "__main__":
    L = []
    lock_con = threading.Condition()
    threads = []
    for i in range(5):
        threads.append(Producer())
    threads.append(Consumer())
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print('---- end ----')

#运行结果:
生产者 Thread-1 :Append63 []
生产者 Thread-2 :Append66 [63]
生产者 Thread-3 :Append20 [63, 66]
生产者 Thread-4 :Append83 [63, 66, 20]
生产者 Thread-5 :Append2 [63, 66, 20, 83]
生产者 Thread-4 :Append26 []
消费者 Thread-6 :Delete26 [26]
生产者 Thread-2 :Append21 []
生产者 Thread-3 :Append71 [21]
生产者 Thread-1 :Append19 [21, 71]
生产者 Thread-5 :Append100 [21, 71, 19]
生产者 Thread-1 :Append96 []
消费者 Thread-6 :Delete96 [96]
........

同步条件

条件同步和条件变量同步差不多意思，只是少了锁功能，因为条件同步设计于不访问共享资源的条件环境。event=threading.Event()：条件环境对象，初始值为False；

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


import threading, time

class Boss(threading.Thread):
    def run(self):
        print("BOSS: 今晚大家加班")
        event.isSet() or event.set()
        time.sleep(5)
        print("BOSS: 大家可以下班了")
        event.isSet() or event.set()


class Worker(threading.Thread):
    def run(self):
        event.wait()
        print("Worker: 唉。。。。")
        time.sleep(0.25)
        event.clear()
        event.wait()
        print("Worker: Great!")


if __name__ == "__main__":
    event = threading.Event()
    threads = []
    for i in range(5):
        threads.append(Worker())
    threads.append(Boss())
    for t in threads:
        t.start()
    for t in threads:
        t.join()

#运行结果:
BOSS: 今晚大家加班
Worker: 唉。。。。
Worker: 唉。。。。
Worker: 唉。。。。
Worker: 唉。。。。
Worker: 唉。。。。
BOSS: 大家可以下班了
Worker: Great!
Worker: Great!
Worker: Great!
Worker: Great!
Worker: Great!

列队

q = Queue.Queue(maxsize = 10) 创建一个“队列”对象。Queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。

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

q.get([block[, timeout]])方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常，timeout等待时间。

q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get_nowait() 相当q.get(False)
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作


import queue

d = queue.Queue()

d.put('1')
d.put('2')
d.put('3')

print(d.get())
print(d.get())
print(d.get())
print(d.get())
print(d.get(0))

# 运行结果:
1
2
3
报错：
queue.Empty

线程操作列表是不安全的。

import threading, time

li = [1, 2, 3, 4, 5]

def pri():
    while li:
        a = li [-1]
        print(a)
        time.sleep(1)
        try:
            li.remove(a)
        except:
            print('-----', a)
t1 = threading.Thread(target=pri, args=())
t1.start()
t2 = threading.Thread(target=pri, args=())
t2.start()

# 运行结果:
5
5
4
----- 5
4
3
----- 4
3
2
----- 3
2
1
----- 2
1
----- 1


import threading, queue
from time import sleep
from random import randint

class Production(threading.Thread):
    def run(self):
        while True:
            r = randint(0, 100)
            q.put(r)
            print("生产出来 %s 号包子" %r)
            sleep(1)

class Proces(threading.Thread):
    def run(self):
        while True:
            re = q.get()
            print('吃掉 %s号包子' %re)

if __name__ == '__main__':
    q = queue.Queue(10)
    threads = [Production(),Production(),Production(),Proces()]
    for t in threads:
        t.start()

# 运行结果:
生产出来 94 号包子
生产出来 13 号包子
生产出来 79 号包子
吃掉 94号包子
吃掉 13号包子
吃掉 79号包子
生产出来 43 号包子
吃掉 43号包子
生产出来 32 号包子
吃掉 32号包子
......

条件变量同步

同步条件

列队

Java synchronized同步关键字背后的原理与源码实现

Android进阶之音视频技术

热门文章