Consumer Application Skeleton (Python)

2015-06-30T03:24:07-07:00

Python recipe 579074 by Vovan (application, consumer, daemon, framework). Revision 2.

Consumer Application Skeleton

This is very basic skeleton for data processing application implementing consumer pattern:

while is_running():
    task = get_next_task_from_queue()
    if task:
        submit_task_for_processing(task)
    else:
        sleep_for_a_moment()

Here's an example:

class ExampleApp(ConsumerAppBase):

    def _get_next_task(self):

        # Get next task from the queue.
        return self._queue.next()

    def _run_task(self, task):

        # This code's being executed in separate worker thread of
        # ThreadPoolExecutor
        return task / 2

    def _on_task_done(self, task, future):

        # Once worker thread finished - task results are available
        # in _on_task_done() callback as a concurrent.futures.Future object.
        self._log.info('Task done. Result: %s', future.result())

Queue with tagged items (Python)

2009-01-28T07:23:40-08:00

Python recipe 576632 by Jirka Vejrazka (consumer, producer, queue, tag). Revision 2.

I needed multiple consumers to retrieve data from a queue fed by one producer. Could not find a good working code for that, so I implemented my own queue. Docstring should describe how this works.

Two notes: 1) my code uses multiprocessing code, but in this module, the Lock and Condition could be easily replaced with the same objects from the threading module 2) the attached test uses syntax for "nose" testing package, I did not convert it to doctest or UnitTest.

Latest recipes tagged "consumer"

Consumer Application Skeleton (Python)

Consumer Application Skeleton

Queue with tagged items (Python)