# 🐰amqp-worker
English | [简体中文](https://git.loom.run/Coder/amqp-worker/src/branch/master/README_zh.md)
amqp-worker is a Python-based multi-threaded RabbitMQ consumer framework. It allows you to consume messages more efficiently and stably.
## Features
- Batch consumption: process messages in batches, improve consumption efficiency.
- Automatic reconnection: when RabbitMQ service disconnects, amqp-worker will automatically reconnect, ensuring uninterrupted consumption.
- Customizable consumption mode: freely decide to use multi-threading and coroutines in the consumption function.
- Configurable message acknowledgment mode: support automatic acknowledgment and manual acknowledgment modes, configure according to your consumption needs.
- Configurable exception handling: support global configuration of message exception consumption mode, re-enter queue, re-insert, consume message.
## Installation
You can use pip tool to install amqp-worker:
```
pip install amqp-workers
```
## Usage
First, you need to import the amqp_worker module in your Python code:
```python
from amqpworker.app import App
```
Then, you need to instantiate an App object, and the App object depends on the AMQPConnection object:
```python
from amqpworker.connections import AMQPConnection
amqp_conn = AMQPConnection(hostname='127.0.0.1', username='guest', password='guest', port=5672)
app = App(connections=[amqp_conn])
```
Next, you need to define the consumption function:
```python
@app.amqp.consume(
['test'],
options=AMQPRouteOptions(bulk_size=1024 * 8, bulk_flush_interval=2)
)
def _handler(msgs: List[RabbitMQMessage]):
print(f"Recv {len(msgs)} {datetime.now().isoformat()}")
```
In the above code we give the consumption function a decorator, giving the consumption queue, the number of consumption per batch, it is worth noting that the parameter type of the consumption function is `List[RabbitMQMessage]`
Finally, just call the `run` method to start consuming:
```python
app.run()
```
## Example code
Below is a simple example code that will consume messages from a queue named `test`:
```python
from datetime import datetime
from typing import List
from amqpworker.app import App
from amqpworker.connections import AMQPConnection
from amqpworker.rabbitmq import RabbitMQMessage
from amqpworker.routes import AMQPRouteOptions
amqp_conn = AMQPConnection(hostname='127.0.0.1', username='guest', password='guest', port=5672)
app = App(connections=[amqp_conn])
@app.amqp.consume(
['test'],
options=AMQPRouteOptions(bulk_size=1024 * 8, bulk_flush_interval=2)
)
def _handler(msgs: List[RabbitMQMessage]):
print(f"Recv {len(msgs)} {datetime.now().isoformat()}")
app.run()
```
## Contributors
- [@JimZhang](https://git.loom.run/zzl221000)
## License
amqp-worker uses MIT license. Please refer to LICENSE file for details.
Raw data
{
"_id": null,
"home_page": "https://git.loom.run/Coder/amqp-worker",
"name": "amqp-workers",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "amqp,rabbitmq",
"author": "JimZhang",
"author_email": "zzl22100048@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/69/cf/a8ded93b7a1b5a6695b6022ecbe3b3990a7bf87102e4f7b7809c3b347373/amqp_workers-0.1.5.tar.gz",
"platform": null,
"description": "# \ud83d\udc30amqp-worker\n\nEnglish | [\u7b80\u4f53\u4e2d\u6587](https://git.loom.run/Coder/amqp-worker/src/branch/master/README_zh.md)\n\n\namqp-worker is a Python-based multi-threaded RabbitMQ consumer framework. It allows you to consume messages more efficiently and stably.\n\n## Features\n\n- Batch consumption: process messages in batches, improve consumption efficiency.\n- Automatic reconnection: when RabbitMQ service disconnects, amqp-worker will automatically reconnect, ensuring uninterrupted consumption.\n- Customizable consumption mode: freely decide to use multi-threading and coroutines in the consumption function.\n- Configurable message acknowledgment mode: support automatic acknowledgment and manual acknowledgment modes, configure according to your consumption needs.\n- Configurable exception handling: support global configuration of message exception consumption mode, re-enter queue, re-insert, consume message.\n\n## Installation\n\nYou can use pip tool to install amqp-worker:\n\n```\npip install amqp-workers\n```\n\n## Usage\n\nFirst, you need to import the amqp_worker module in your Python code:\n\n```python\nfrom amqpworker.app import App\n```\n\nThen, you need to instantiate an App object, and the App object depends on the AMQPConnection object:\n\n```python\nfrom amqpworker.connections import AMQPConnection\namqp_conn = AMQPConnection(hostname='127.0.0.1', username='guest', password='guest', port=5672)\n\napp = App(connections=[amqp_conn])\n```\n\n\n\nNext, you need to define the consumption function:\n\n```python\n@app.amqp.consume(\n ['test'],\n options=AMQPRouteOptions(bulk_size=1024 * 8, bulk_flush_interval=2)\n)\ndef _handler(msgs: List[RabbitMQMessage]):\n print(f\"Recv {len(msgs)} {datetime.now().isoformat()}\")\n```\n\n\nIn the above code we give the consumption function a decorator, giving the consumption queue, the number of consumption per batch, it is worth noting that the parameter type of the consumption function is `List[RabbitMQMessage]`\n\nFinally, just call the `run` method to start consuming:\n\n```python\napp.run()\n```\n\n## Example code\n\nBelow is a simple example code that will consume messages from a queue named `test`:\n\n```python\nfrom datetime import datetime\nfrom typing import List\n\nfrom amqpworker.app import App\nfrom amqpworker.connections import AMQPConnection\nfrom amqpworker.rabbitmq import RabbitMQMessage\nfrom amqpworker.routes import AMQPRouteOptions\n\namqp_conn = AMQPConnection(hostname='127.0.0.1', username='guest', password='guest', port=5672)\napp = App(connections=[amqp_conn])\n\n@app.amqp.consume(\n ['test'],\n options=AMQPRouteOptions(bulk_size=1024 * 8, bulk_flush_interval=2)\n)\ndef _handler(msgs: List[RabbitMQMessage]):\n print(f\"Recv {len(msgs)} {datetime.now().isoformat()}\")\n\napp.run()\n\n```\n\n## Contributors\n\n- [@JimZhang](https://git.loom.run/zzl221000)\n\n## License\n\namqp-worker uses MIT license. Please refer to LICENSE file for details.",
"bugtrack_url": null,
"license": "MIT",
"summary": "amqp-worker is a Python-based multi-threaded RabbitMQ consumer framework",
"version": "0.1.5",
"project_urls": {
"Homepage": "https://git.loom.run/Coder/amqp-worker",
"Repository": "https://git.loom.run/Coder/amqp-worker"
},
"split_keywords": [
"amqp",
"rabbitmq"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "80b912eb990fb98fb76ea143047c779b6ef9db8d4a10d6e01213553eb024ce94",
"md5": "7bd01483cb349f9fda6cc31853414547",
"sha256": "e48b241eee0f7e929c6e36922c88a89b0f2df4c5fcd44bfe6164410df4f23054"
},
"downloads": -1,
"filename": "amqp_workers-0.1.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7bd01483cb349f9fda6cc31853414547",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 28476,
"upload_time": "2024-01-22T08:46:13",
"upload_time_iso_8601": "2024-01-22T08:46:13.157626Z",
"url": "https://files.pythonhosted.org/packages/80/b9/12eb990fb98fb76ea143047c779b6ef9db8d4a10d6e01213553eb024ce94/amqp_workers-0.1.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "69cfa8ded93b7a1b5a6695b6022ecbe3b3990a7bf87102e4f7b7809c3b347373",
"md5": "6e1ebc67c3af2458a8387306d4cb2b79",
"sha256": "3260555d1d851a4b04b2e70603824f5a0266ed13bd8f30e6a0516bb9be5edeef"
},
"downloads": -1,
"filename": "amqp_workers-0.1.5.tar.gz",
"has_sig": false,
"md5_digest": "6e1ebc67c3af2458a8387306d4cb2b79",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 20180,
"upload_time": "2024-01-22T08:46:14",
"upload_time_iso_8601": "2024-01-22T08:46:14.375891Z",
"url": "https://files.pythonhosted.org/packages/69/cf/a8ded93b7a1b5a6695b6022ecbe3b3990a7bf87102e4f7b7809c3b347373/amqp_workers-0.1.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-01-22 08:46:14",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "amqp-workers"
}
JSON