Task routing

[Timaqf]

does dramatiq has anything like celery about task routing.

I didn't find any api to send task with queue_name, but I can build message myself to send task to specific queue.

 Message(
            queue_name="only_i_can_receive",
            actor_name="count_words",
            args=("https://www.google.com", ),
        )
broker.enqueue(message, delay=delay)

but i can't start worker only consumer for specific queue like celery.

celery worker tasks -Q tasks.to.nodes.1

[Timaqf]

in celery, I can start worker with different queue name, so I can send task to specific worker with different queue name.

but I can't control Worker.consumers from dramatiq-gevent script, maybe I didn't see any api to do this.

[Timaqf]

I find a hacking way, I can remove _WorkerMiddleware at here.

dramatiq/dramatiq/worker.py

Line 59 in 2855d1a

self.broker.add_middleware(worker_middleware)

and use worker._add_consumer(queue_name) after here.

dramatiq/dramatiq/__main__.py

Line 159 in 2855d1a

worker = Worker(broker, worker_threads=args.threads)

somethings like

parser.add_argument(
    "--queues", "-Q", default=["default"], nargs="*", type=str,
    help="the queues of worker to consuming (default: default)"
)

for queue_name in args.queues:
    worker.logger.debug("Adding consumer for queue %r.", queue_name)
    worker._add_consumer(queue_name)

but this is conflict with current queue_name check.

dramatiq/dramatiq/actor.py

Line 55 in 620f460

if not _queue_name_re.fullmatch(queue_name):

[Bogdanp]

Yep, something like that ought to work. This use case isn't supported right now, but it's something I can add in the future.

Can you tell me more about your exact use case here, @Timaqf? Would actor priorities not be a good fit?

[Timaqf]

@Bogdanp I want different worker to do different work in same app depends on work's args and kwargs.

[Timaqf]

@Bogdanp
somethings like server A only do priority HIGH job, server B only do priority LOW job. and I can decide job priority depends on job's args and kwargs.

[Bogdanp]

Gotcha. Right now you can use actor priorities for something like that but the disadvantage is priorities are per-actor and workers are homogeneous. I'll add something like your proposed enhancement soon.

[Timaqf]

nice man!

[Bogdanp]

Apologies for the delay, @Timaqf. This is now in master and I'll cut a release tomorrow!

[Timaqf]

thanks.
but I read the code in 0.20.0, the --queues seems like white queues, it needs to be declared in actor's decoration, so it's in the code not in the command line.

like below

@dramatiq.actor(queue_name="low")
def count_words(*args):
    return 0

@dramatiq.actor(queue_name="high")
def count_words(*args):
    return 0

so I can start the worker only do high jobs like dramatiq count_words --queues high.
the question is I need to define function twice if I have two queues. and also the queues need to be declared in the code before the starting.

if do some changes in worker_process, declard the queue before the worker starts.

https://github.com/Bogdanp/dramatiq/blob/master/dramatiq/__main__.py#L165

        worker = Worker(broker, queues=args.queues, worker_threads=args.threads)
        for queue_name in args.queues:
            worker.broker.declare_queue(queue_name)
        worker.start()

I can write the below code to do the same jobs.

@dramatiq.actor(queue_name="count_words_queue_default")
def count_words(*args):
    return 0

the starts like dramatiq count_words --queues count_words_queue_high.

let me known what you think.

[Bogdanp]

I think I see what you're saying. Unfortunately, the fact that each actor is bound to a single queue is a core part of the design and I don't think that should change. You could achieve what you're currently doing with modifying __main__ by writing a middleware and declaring the queues in its after_worker_boot hook.