基于event 实现的线程安全的优先队列(python实现)

Posted 2020-10-09

event 事件是个很不错的线程同步，以及线程通信的机制，在python的许多源代码中都基于event实现了很多的线程安全，支持并发，线程通信的库

对于优先队列的堆实现，请看《python下实现二叉堆以及堆排序》, python的event请看<python lock, semaphore, event实现线程同步>, 其实主要注意event几个方法的用法, 以及多线程访问下的程序的逻辑顺序, 在相关代码段放好event的几个方法就好了。自己去理解下, 不难。看下源码吧：

import heapq
import threading
# import time


class Item:
    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return ‘Item({!r})‘.format(self.name)


class PriorityQueue:
    def __init__(self):
        self._queue = []
        self._index = 0
        self._event = threading.Event()

    def push(self, item, priority):
        if len(self._queue) != 0:
            self._event.clear()
        while not self._event.is_set():
            self._event.set()
            heapq.heappush(self._queue, (-priority, self._index, item))  # 存入一个三元组, 默认构造的是小顶堆
            self._index += 1
            # self._event.set()

    def pop(self):
        if len(self._queue) != 0:
            self._event.set()
        while self._event.is_set():
            self._event.wait()
            x = heapq.heappop(self._queue)[-1]  # 逆序输出
            self._event.clear()
            return x


def test1(p, item, index):
    for i in range(3):
        p.push(Item(item), index)


def test2(p):
    for i in range(3):
        print(p.pop())


if __name__ == ‘__main__‘:
    p = PriorityQueue()

    t1 = threading.Thread(target=test1, args=(p, ‘foo‘, 1))
    t3 = threading.Thread(target=test1, args=(p, ‘bar‘, 2))
    t4 = threading.Thread(target=test1, args=(p, ‘Ryan‘, 28))

    t2 = threading.Thread(target=test2, args=(p,))
    t5 = threading.Thread(target=test2, args=(p,))
    t6 = threading.Thread(target=test2, args=(p,))

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    t3.start()
    t5.start()

    t3.join()
    t5.join()

    t4.start()
    t6.start()

    t4.join()
    t6.join()

建议去看看python的socket源码以及queue.py等源码这些涉及到了多线程访问的库都是基于这些线程同步机制实现的, 你可以仿写下, 毕竟人家写的更专业.