Python rq 模块，Worker() 实例源码

def worker(config_uri):
    """ Console entry script that starts a worker process
    """
    # TODO: import spacy's model to share it between workers

    pyramid_env = bootstrap(config_uri)

    # this conflicts with normal worker output
    # TODO: solve logging for the console
    # Setup logging to allow log output from command methods
    # from pyramid.paster import setup_logging
    # setup_logging(config_uri)

    try:
        qs = ['default']
        conn = redis_connection()
        with Connection(conn):
            w = Worker(qs)
            w.work()
    finally:
        pyramid_env['closer']()

def __init__(self, queues=None, *args, **kwargs):
        u'''
        Constructor.

        Accepts the same arguments as the constructor of
        ``rq.worker.Worker``. However, the behavior of the ``queues``
        parameter is different.

        :param queues: The job queue(s) to listen on. Can be a string
            with the name of a single queue or a list of queue names.
            If not given then the default queue is used.
        '''
        queues = queues or [DEFAULT_QUEUE_NAME]
        queues = [get_queue(q) for q in ensure_list(queues)]
        rq.worker.logger.setLevel(logging.INFO)
        super(Worker, self).__init__(queues, *args, **kwargs)

def perform_job(self, *args, **kwargs):
        result = super(Worker, self).perform_job(*args, **kwargs)
        # rq.Worker.main_work_horse does a hard exit via os._exit directly
        # after its call to perform_job returns. Hence here is the correct
        # location to clean up.
        try:
            meta.Session.remove()
        except Exception:
            log.exception(u'Error while closing database session')
        try:
            meta.engine.dispose()
        except Exception:
            log.exception(u'Error while disposing database engine')
        return result

# adapted from ckanext.datastore.backend.postgres

def worker():
    with Connection(Redis("jobqueue.local")):
        qs = sys.argv[1:] or ['default']
        print("foo")
        w = Worker(qs)
        w.work()

def get_worker(*queue_names):
    """
    Returns a RQ worker for all queues or specified ones.
    """
    queues = get_queues(*queue_names)
    return Worker(queues,
                  connection=queues[0].connection,
                  exception_handlers=get_exception_handlers() or None)

def main():
    """
        Init workers
    """
    parser = argparse.ArgumentParser()
    parser.add_argument('--queues', nargs='+', type=unicode, dest='queues', required=True)
    args = parser.parse_args()

    with Connection(connection=StrictRedis(**settings.REDIS)):

        qs = map(Queue, args.queues) or [Queue()]
        worker = Worker(qs)
        worker.work()

def main():
    with Connection(redis_connection):
        worker = Worker(Queue('default'))
        worker.work()

def request_stop(self, *args, **kwargs):
        """When SIGINT is sent to the worker (eg, if the Supervisor process
        group is restarted), immediately fail the running job and stop the
        worker. This avoids a scenario in which the worker gets shut down but
        not unregistered in Redis, causing it to get "marooned" in the admin.
        """
        job = self.get_current_job()

        if job:
            self.handle_job_failure(job)
            self.failed_queue.quarantine(job, 'Worker shutdown.')

        self.request_force_stop(*args, **kwargs)

def main(config):
    global worker_config
    worker_config = config

    listen = config["listen"].values()

    queue_dsn = config["queue"]["dsn"]

    conn = redis.from_url(queue_dsn)

    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

def main(config=None):

    listen = config["listen"].values()

    queue_dsn = config["queue"]["dsn"]

    conn = redis.from_url(queue_dsn)

    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

def register_birth(self, *args, **kwargs):
        result = super(Worker, self).register_birth(*args, **kwargs)
        names = [remove_queue_name_prefix(n) for n in self.queue_names()]
        names = u', '.join(u'"{}"'.format(n) for n in names)
        log.info(u'Worker {} (PID {}) has started on queue(s) {} '.format(
                 self.key, self.pid, names))
        return result

def execute_job(self, job, *args, **kwargs):
        # We shut down all database connections and the engine to make sure
        # that they are not shared with the child process and closed there
        # while still being in use in the main process, see
        #
        #   https://github.com/ckan/ckan/issues/3365
        #
        # Note that this rolls back any non-committed changes in the session.
        # Both `Session` and `engine` automatically re-initialize themselve
        # when they are used the next time.
        log.debug(u'Disposing database engine before fork')
        meta.Session.remove()
        meta.engine.dispose()

        # The original implementation performs the actual fork
        queue = remove_queue_name_prefix(job.origin)
        log.info(u'Worker {} starts job {} from queue "{}"'.format(
                 self.key, job.id, queue))

        # HACK
        # for plugin in plugins.PluginImplementations(plugins.IForkObserver):
        #     plugin.before_fork()
        _dispose_engines()

        result = super(Worker, self).execute_job(job, *args, **kwargs)
        log.info(u'Worker {} has finished job {} from queue "{}"'.format(
                 self.key, job.id, queue))

        return result

def register_death(self, *args, **kwargs):
        result = super(Worker, self).register_death(*args, **kwargs)
        log.info(u'Worker {} (PID {}) has stopped'.format(self.key, self.pid))
        return result

def handle_exception(self, job, *exc_info):
        log.exception(u'Job {} on worker {} raised an exception: {}'.format(
                      job.id, self.key, exc_info[1]))
        return super(Worker, self).handle_exception(job, *exc_info)

def main():
    with Connection(Redis(settings['redis server'])):
        qs = ['default']
        w = Worker(qs)
        w.work()

def runworker(app):

    REDIS_HOST = app.config['REDIS_HOST']
    REDIS_PORT = app.config['REDIS_PORT']
    REDIS_DB = app.config['REDIS_DB']
    QUEUES = app.config['QUEUES']

    redis_conn = Connection(Redis(REDIS_HOST,
                                  REDIS_PORT,
                                  REDIS_DB))
    with redis_conn:
        w = Worker(QUEUES)
        w.work()

def launch_rq_worker() -> None:
    """
    Blocking function to launch a worker using Python RQ's internal API
    """
    with Connection():
        w = Worker(
            get_available_rq_worker_name()
        )

        w.work()

def run_worker():
    """Initializes a slim rq task queue."""
    listen = ['default']
    conn = Redis(
        host=app.config['RQ_DEFAULT_HOST'],
        port=app.config['RQ_DEFAULT_PORT'],
        db=0,
        password=app.config['RQ_DEFAULT_PASSWORD'])

    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

def work():
    print("Hello from the worker side.")
    with Connection(REDIS):
        worker = Worker(map(Queue, QUEUES))
        worker.work()

def start_worker(queue_name):
    print "starting worker '{}'...".format(queue_name)

    with Connection(redis_rq_conn):
        worker = Worker(Queue(queue_name), exc_handler=failed_job_handler)
        worker.work()

def run_worker():
    """Initializes a slim rq task queue."""
    listen = ['default']
    conn = Redis(
        host=app.config['RQ_DEFAULT_HOST'],
        port=app.config['RQ_DEFAULT_PORT'],
        db=0,
        password=app.config['RQ_DEFAULT_PASSWORD'])

    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()