python类queue()的实例源码

def __init__(self, app, pool_size=30):
        self.task_queue = gevent.queue.Queue()
        self.pool_size = pool_size
        self.app = app

def test_queue2(self):
        """?????size?????get/set????????"""
        _log.info('test_queue2222222222')
        task_queue = Queue(3)
        def worker(name):
            try:
                while True:
                    task = task_queue.get(timeout=1) # decrements queue size by 1
                    print('Worker %s got task %s' % (name, task))
                    gevent.sleep(0)
            except Empty:
                print('Quitting time!')

        def boss():
            """
            Boss will wait to hand out work until a individual worker is
            free since the maxsize of the task queue is 3.
            """

            for i in xrange(1,10):
                task_queue.put(i)
            print('Assigned all work in iteration 1')

            for i in xrange(10,20):
                task_queue.put(i)
            print('Assigned all work in iteration 2')

        gevent.joinall([
            gevent.spawn(boss),
            gevent.spawn(worker, 'steve'),
            gevent.spawn(worker, 'john'),
            gevent.spawn(worker, 'bob'),
        ])

def worker(n):
    try:
        while True:
            task = tasks.get(timeout=1) # decrements queue size by 1
            print('Worker %s got task %s' % (n, task))
            gevent.sleep(0) # yielding????Greenlet???????
    except Empty:
        print('Quitting time!')

def boss():
    """
    Boss will wait to hand out work until a individual workeworker 
    free since the maxsize of the task queue is 3.
    """
    for i in range(1,10):
        tasks.put(i)
    print('Assigned all work in iteration 1')
    for i in range(10,20):
        tasks.put(i)
    print('Assigned all work in iteration 2')

def _create_greenlet_worker(self, queue):
        def greenlet_worker():
            while True:
                try:
                    func = queue.get()
                    if func is _STOP:
                        break
                    func()
                except Empty:
                    continue
                except Exception as exc:
                    log.warning("Exception in worker greenlet")
                    log.exception(exc)
        return gevent.spawn(greenlet_worker)

def start(self):
        """Start the greenlet workers."""
        with self._state_change:
            if self._running:
                return

            self._running = True

            # Spawn our worker greenlets, we have
            # - A callback worker for watch events to be called
            for queue in (self.callback_queue,):
                w = self._create_greenlet_worker(queue)
                self._workers.append(w)
            python2atexit.register(self.stop)

def stop(self):
        """Stop the request processor."""
        shared = self.shared
        self.shared = None
        log.info("RequestHandler.stop: about to flush requests queue")
        shared.requests.join()
        shared.ending.set()

def _start_thread(self):
        """Run the request processor"""
        # We pass a direct reference to `shared` into the worker, to avoid
        # that thread holding a ref to `self`, which would prevent GC.  A
        # previous version of this used a weakref to `self`, but would
        # potentially abort the thread before the requests queue was empty
        shared = self.shared

        def worker():
            try:
                while not shared.ending.is_set():
                    try:
                        # set a timeout so we check `ending` every so often
                        task = shared.requests.get(timeout=1)
                    except Empty:
                        continue
                    try:
                        shared.connection.request(task.request)
                        if task.future:
                            res = shared.connection.response()
                            task.future.set_response(res)
                    except Exception as e:
                        if task.future:
                            task.future.set_error(e)
                    finally:
                        shared.requests.task_done()
                log.info("RequestHandler worker: exiting cleanly")
            except:
                # deal with interpreter shutdown in the same way that
                # python 3.x's threading module does, swallowing any
                # errors raised when core modules such as sys have
                # already been destroyed
                if _sys is None:
                    return
                raise

        name = "pykafka.RequestHandler.worker for {}:{}".format(
            self.shared.connection.host, self.shared.connection.port)
        return self.handler.spawn(worker, name=name)

def __call__(self, environ, start_response):
        self.environ = environ

        uwsgi.websocket_handshake()

        self._req_ctx = None
        if hasattr(uwsgi, 'request_context'):
            # uWSGI >= 2.1.x with support for api access across-greenlets
            self._req_ctx = uwsgi.request_context()
        else:
            # use event and queue for sending messages
            from gevent.event import Event
            from gevent.queue import Queue
            from gevent.select import select
            self._event = Event()
            self._send_queue = Queue()

            # spawn a select greenlet
            def select_greenlet_runner(fd, event):
                """Sets event when data becomes available to read on fd."""
                while True:
                    event.set()
                    try:
                        select([fd], [], [])[0]
                    except ValueError:
                        break
            self._select_greenlet = gevent.spawn(
                select_greenlet_runner,
                uwsgi.connection_fd(),
                self._event)

        self.app(self)

def wait(self):
        """Waits and returns received messages.
        If running in compatibility mode for older uWSGI versions,
        it also sends messages that have been queued by send().
        A return value of None means that connection was closed.
        This must be called repeatedly. For uWSGI < 2.1.x it must
        be called from the main greenlet."""
        while True:
            if self._req_ctx is not None:
                try:
                    msg = uwsgi.websocket_recv(request_context=self._req_ctx)
                except IOError:  # connection closed
                    return None
                return self._decode_received(msg)
            else:
                # we wake up at least every 3 seconds to let uWSGI
                # do its ping/ponging
                event_set = self._event.wait(timeout=3)
                if event_set:
                    self._event.clear()
                    # maybe there is something to send
                    msgs = []
                    while True:
                        try:
                            msgs.append(self._send_queue.get(block=False))
                        except gevent.queue.Empty:
                            break
                    for msg in msgs:
                        self._send(msg)
                # maybe there is something to receive, if not, at least
                # ensure uWSGI does its ping/ponging
                try:
                    msg = uwsgi.websocket_recv_nb()
                except IOError:  # connection closed
                    self._select_greenlet.kill()
                    return None
                if msg:  # message available
                    return self._decode_received(msg)

def received_message(self, message):
        """
        Override the base class to store the incoming message
        in the `messages` queue.
        """
        self.messages.put(copy.deepcopy(message))

def closed(self, code, reason=None):
        """
        Puts a :exc:`StopIteration` as a message into the
        `messages` queue.
        """
        # When the connection is closed, put a StopIteration
        # on the message queue to signal there's nothing left
        # to wait for
        self.messages.put(StopIteration)

def receive(self):
        """
        Returns messages that were stored into the
        `messages` queue and returns `None` when the
        websocket is terminated or closed.
        """
        # If the websocket was terminated and there are no messages
        # left in the queue, return None immediately otherwise the client
        # will block forever
        if self.terminated and self.messages.empty():
            return None
        message = self.messages.get()
        if message is StopIteration:
            return None
        return message

def received_message(self, message):
        """
        Override the base class to store the incoming message
        in the `messages` queue.
        """
        self.messages.put(copy.deepcopy(message))

def closed(self, code, reason=None):
        """
        Puts a :exc:`StopIteration` as a message into the
        `messages` queue.
        """
        # When the connection is closed, put a StopIteration
        # on the message queue to signal there's nothing left
        # to wait for
        self.messages.put(StopIteration)

def receive(self):
        """
        Returns messages that were stored into the
        `messages` queue and returns `None` when the
        websocket is terminated or closed.
        """
        # If the websocket was terminated and there are no messages
        # left in the queue, return None immediately otherwise the client
        # will block forever
        if self.terminated and self.messages.empty():
            return None
        message = self.messages.get()
        if message is StopIteration:
            return None
        return message

def wait_for_action(self, timeout=3600, raise_on_hangup=False):
        """
        Wait until an action is over
        and return action event.
        """
        self.log.debug("wait for action start")
        try:
            event = self._action_queue.get(timeout=timeout)
            self.log.debug("wait for action end %s" % str(event))
            if raise_on_hangup is True and self.has_hangup():
                self.log.warn("wait for action call hung up !")
                raise RESTHangup()
            return event
        except gevent.queue.Empty:
            if raise_on_hangup is True and self.has_hangup():
                self.log.warn("wait for action call hung up !")
                raise RESTHangup()
            self.log.warn("wait for action end timed out!")
            return Event()


    # In order to "block" the execution of our service until the
    # command is finished, we use a synchronized queue from gevent
    # and wait for such event to come. The on_channel_execute_complete
    # method will put that event in the queue, then we may continue working.
    # However, other events will still come, like for instance, DTMF.

def disconnect(self):
        # Prevent command to be stuck while waiting response
        try:
            self._action_queue.put_nowait(Event())
        except gevent.queue.Full:
            pass
        self.log.debug('Releasing Connection ...')
        super(PlivoOutboundEventSocket, self).disconnect()
        self.log.debug('Releasing Connection Done')

def __init__(self, other, *args, **kwargs):
        from gevent.queue import Queue

        super(MemoryBufferedPlayable, self).__init__(*args, **kwargs)
        self.frames = Queue()
        self.other = other
        gevent.spawn(self._buffer)

def test_worker(self, gen_id_mock, call_mock):
        call_mock.side_effect = TestContestManager._valid_subprocess_call
        ContestManager.tasks["poldo"] = { "generator": "/gen", "validator": "/val" }

        with patch("src.logger.Logger.error", side_effect=TestContestManager._stop_worker_loop):
            with patch("gevent.queue.Queue.put", side_effect=NotImplementedError("Stop loop")):
                with self.assertRaises(NotImplementedError) as ex:
                    ContestManager.worker("poldo")