You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Copy file name to clipboardExpand all lines: README.md
+45-33Lines changed: 45 additions & 33 deletions
Display the source diff
Display the rich diff
Original file line number
Diff line number
Diff line change
@@ -9,13 +9,14 @@ This library provides you with aio-pika broker for taskiq.
9
9
Features:
10
10
- Supports delayed messages using dead-letter queues or RabbitMQ delayed message exchange plugin.
11
11
- Supports message priorities.
12
+
- Supports multiple queues and custom routing.
12
13
13
14
Usage example:
14
15
15
16
```python
16
17
from taskiq_aio_pika import AioPikaBroker
17
18
18
-
broker = AioPikaBroker()
19
+
broker = AioPikaBroker(...)
19
20
20
21
@broker.task
21
22
asyncdeftest() -> None:
@@ -27,12 +28,23 @@ async def test() -> None:
27
28
28
29
### Default delays
29
30
30
-
To send delayed message, you have to specify delay label. You can do it with `task` decorator, or by using kicker.
31
+
To send delayed message, you need to specify queue for delayed messages. You can do it by passing `delay_queue` parameter to the broker. For example:
32
+
33
+
```python
34
+
from taskiq_aio_pika import AioPikaBroker, Queue, QueueType
35
+
36
+
broker = AioPikaBroker(
37
+
...,
38
+
delay_queue=Queue(name="taskiq.delay_queue"),
39
+
)
40
+
```
41
+
42
+
After that you have to specify delay label. You can do it with `task` decorator, or by using kicker.
31
43
32
44
In this type of delay we are using additional queue with `expiration` parameter. After declared time message will be deleted from `delay` queue and sent to the main queue. For example:
33
45
34
46
```python
35
-
broker = AioPikaBroker()
47
+
broker = AioPikaBroker(...)
36
48
37
49
@broker.task(delay=3)
38
50
asyncdefdelayed_task() -> int:
@@ -86,13 +98,12 @@ async def main():
86
98
## Priorities
87
99
88
100
You can define priorities for messages using `priority` label. Messages with higher priorities are delivered faster.
89
-
But to use priorities you need to define `max_priority` of the main queue, by passing `max_priority` parameter in broker's init. This parameter sets maximum priority for the queue and declares it as the priority queue.
90
101
91
102
Before doing so please read the [documentation](https://www.rabbitmq.com/priority.html#behaviour) about what
*`url` - url to rabbitmq. If None, "amqp://guest:guest@localhost:5672" is used.
119
-
*`result_backend` - custom result backend.
120
-
*`task_id_generator` - custom task_id genertaor.
121
-
*`exchange_name` - name of exchange that used to send messages.
122
-
*`exchange_type` - type of the exchange. Used only if `declare_exchange` is True.
123
-
*`queue_name` - queue that used to get incoming messages.
124
-
*`routing_key` - that used to bind that queue to the exchange.
125
-
*`declare_exchange` - whether you want to declare new exchange if it doesn't exist.
126
-
*`max_priority` - maximum priority for messages.
127
-
*`delay_queue_name` - custom delay queue name. This queue is used to deliver messages with delays.
128
-
*`dead_letter_queue_name` - custom dead letter queue name.
129
-
This queue is used to receive negatively acknowledged messages from the main queue.
130
-
*`qos` - number of messages that worker can prefetch.
131
-
*`declare_queues` - whether you want to declare queues even on client side. May be useful for message persistence.
132
-
*`declare_queues_kwargs` - see [Custom Queue Arguments](#custom-queue-arguments) for more details.
133
-
134
-
## Custom Queue Arguments
135
-
136
-
You can pass custom arguments to the underlying RabbitMQ queue declaration by using the `declare_queues_kwargs` parameter of `AioPikaBroker`. If you want to set specific queue arguments (such as RabbitMQ extensions or custom behaviors), provide them in the `arguments` dictionary inside `declare_queues_kwargs`.
127
+
You can pass custom arguments to the underlying RabbitMQ queues and exchange declaration by using the `Queue`/`Exchange` classes from `taskiq_aio_pika`. If you used `faststream` before you are probably familiar with this concept.
137
128
138
129
These arguments will be merged with the default arguments used by the broker
139
130
(such as dead-lettering and priority settings). If there are any conflicts, the values you provide will take precedence over the broker's defaults. Example:
140
131
141
132
```python
133
+
from taskiq_aio_pika import AioPikaBroker, Queue, QueueType, Exchange
134
+
from aio_pika.abc import ExchangeType
135
+
142
136
broker = AioPikaBroker(
143
-
declare_queues_kwargs={
144
-
"arguments": {
145
-
"x-message-ttl": 60000, # Set message TTL to 60 seconds
146
-
"x-queue-type": "quorum", # Use quorum queue type
147
-
}
148
-
}
137
+
exchange=Exchange(
138
+
name="custom_exchange",
139
+
type=ExchangeType.TOPIC,
140
+
declare=True,
141
+
durable=True,
142
+
auto_delete=False,
143
+
),
144
+
task_queues=[
145
+
Queue(
146
+
name="custom_queue",
147
+
type=QueueType.CLASSIC,
148
+
declare=True,
149
+
durable=True,
150
+
max_priority=10,
151
+
routing_key="custom_queue",
152
+
)
153
+
]
149
154
)
150
155
```
151
156
152
157
This will ensure that the queue is created with your custom arguments, in addition to the broker's defaults.
158
+
159
+
160
+
## Multiqueue support
161
+
162
+
You can define multiple queues for your tasks. Each queue can have its own routing key and other settings. And your workers can listen to multiple queues (or specific queue) as well.
163
+
164
+
You can check [multiqueue usage example](./examples/topic_with_two_queues.py) in examples folder for more details.
0 commit comments