human/utils/sync_queue.py

#encoding = utf8

import threading
from queue import Queue


class SyncQueue:
    def __init__(self, maxsize, name):
        self._name = name
        self._queue = Queue(maxsize)
        self._condition = threading.Condition()

    def is_empty(self):
        return self._queue.empty()

    def put(self, item):
        with self._condition:
            while self._queue.full():
                # print(self._name, 'put wait')
                self._condition.wait()
            self._queue.put(item)
            self._condition.notify()

    def get(self, timeout=None):
        # 添加超时时间，防止死锁
        with self._condition:
            while self._queue.empty():
                self._condition.wait(timeout=timeout)
                # print(self._name, 'get wait')
                if timeout is not None:
                    break
            item = self._queue.get(block=False)
            self._condition.notify()
            return item

    def clear(self):
        with self._condition:
            while not self._queue.empty():
                self._queue.queue.clear()
            self._condition.notify()

    def size(self):
        return self._queue.qsize()


'''
class SyncQueue:
    def __init__(self, maxsize):
        self._queue = Queue()

    def put(self, item):
        self._queue.put(item)

    def get(self):
        return self._queue.get(block=True, timeout=0.2)

    def clear(self):
        self._queue.queue.clear()

    def size(self):
        return self._queue.qsize()
'''