Python 线程同步和线程优先级队列_python线程优先级

一、线程同步

如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步

使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间

多线程的优势在于可以同时运行多个任务（至少感觉起来是这样）。但是当线程需要共享数据时，可能存在数据不同步的问题。
考虑这样一种情况：一个列表里所有元素都是0，线程"set"从后向前把所有元素改成1，而线程"print"负责从前往后读取列表并打印。
那么，可能线程"set"开始改的时候，线程"print"便来打印列表了，输出就成了一半0一半1，这就是数据的不同步。为了避免这种情况，引入了锁的概念。
锁有两种状态——锁定和未锁定。每当一个线程比如 “set” 要访问共享数据时，必须先获得锁定；如果已经有别的线程比如 “print” 获得锁定了，那么就让线程"set"暂停，也就是同步阻塞；等到线程"print"访问完毕，释放锁以后，再让线程"set"继续。
经过这样的处理，打印列表时要么全部输出0，要么全部输出1，不会再出现一半0一半1的尴尬场面。

import threading
from time import sleep,ctime

class Mythread(threading.Thread):
    def __init__(self,threadID,name,counter):
        super(Mythread, self).__init__()
        self.threadID = threadID
        self.name = name
        self.counter = counter

    def run(self):
        print("开启线程："+self.name)
        # 获取锁，用于线程同步
        threadLock.acquire()
        print_time(self.name,self.counter,3)
        #释放锁，开启下一个线程
        threadLock.release()

def print_time(threadName,delay,counter):
    while counter:
        sleep(delay)
        print("%s : %s" % (threadName,ctime()))
        counter -=1

threadLock = threading.Lock()
threads = []

thread1 = Mythread(1,'Thread-1',1)
thread2 = Mythread(2,'Thread-2',2)
threads.append(thread1)
threads.append(thread2)

# 开启线程
for i in threads:
    i.start()

for i in threads:
    i.join()

print("退出主线程")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

二、线程优先级队列

Python 的 Queue 模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列 PriorityQueue。

这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程间的同步

Queue 模块中的常用方法:

Queue.qsize() 返回队列的大小
Queue.empty() 如果队列为空，返回True,反之False
Queue.full() 如果队列满了，返回True,反之False
Queue.full 与 maxsize 大小对应
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()函数向任务已经完成的队列发送一个信号
Queue.join() 实际上意味着等到队列为空，再执行别的操作
1
2
3
4
5
6
7
8
9
10

import queue
import threading
from time import sleep,ctime

exitflag = 0

class Mythread(threading.Thread):
    def __init__(self,threadID,name,q):
        super(Mythread, self).__init__()
        self.threadID = threadID
        self.name = name
        self.q = q

    def run(self):
        print("开启线程：" + self.name + " at：" + ctime())
        process_data(self.name,self.q)
        print("退出线程：" + self.name)

def process_data(threadName,q):
    while not exitflag:
        queueLock.acquire()
        if not workqueue.empty():
            data = q.get()
            queueLock.release()
            print("%s processing %s" % (threadName,data))
        else:
            queueLock.release()
        sleep(1)

threadList = ["Thread-1","Thread-2","Thread-3"]
nameList = ["One","Two","Three","Four","Five"]
queueLock = threading.Lock()
workqueue = queue.Queue(10)
threads = []
threadID = 1

# 创建新线程

for tName in threadList:
    thread = Mythread(threadID,tName,workqueue)
    thread.start()
    threads.append(thread)
    threadID +=1

# 填充队列
queueLock.acquire()
for word in nameList:
    workqueue.put(word)
queueLock.release()

# 等待队列清空
while not workqueue.empty():
    pass

# 通知线程是时候退出
exitflag = 1

for t in threads:
    t.join()
print("退出主线程")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60