轻量级Python AMQP服务器实现用于开发和模拟?

8 投票
2 回答
3182 浏览
提问于 2025-04-16 12:54

在Django中,运行 ./manage.py runserver 对于开发来说非常方便,这样就不用麻烦去设置和启动一个真正的网络服务器了。

如果你没有使用Django,还是可以很简单地设置一个gunicorn服务器。

那AMQP有没有类似的东西呢?

我不需要一个完整的实现或者很复杂的东西,只要一个简单易安装、方便开发用的就行。要是有个PyPi包就更好了。

Celery不是我想要的。我不需要一个客户端,我想要一个服务器。就像一个迷你版的Python RabbitMq。

2 个回答

2

我之前也有过同样的问题,发现这个领域的信息实在太少了。即使过了这么久,网上几乎找不到轻量级的AMQP服务器。我甚至在Github上找不到什么简单的实现。AMQP这个协议看起来真是个庞然大物。我发现RabbitMQ可能是最轻量的选择了。

最后,我选择了一个基于Docker的解决方案来进行我的集成测试,这个方案可以自动启动和关闭一个RabbitMQ的容器。你需要安装Docker的Python库,当然,你的电脑上还得有Docker服务在运行。我之前就已经在用Docker做其他事情,所以对我来说设置起来并不麻烦;但你可能会有不同的体验。之后,我基本上是这样做的:

import docker

client = docker.from_env()
c = client.containers.run(
    'rabbitmq:alpine',                       # use Alpine Linux build (smaller)
    auto_remove=True,                        # Remove automatically when stopped
    detach=True,                             # Run in daemon mode  
    ports={'5672/tcp' : ('127.0.0.1', None)} # Bind to a random localhost port
)

container = client.containers.get(c.id)      # Re-fetch container for port
port = container.attrs['NetworkSettings']['Ports']['5672/tcp'][0]['HostPort']

# ... Do any set up of the RabbitMQ instance needed on (127.0.0.1:<port>)

# ... Run tests against (127.0.0.1:<port>)

container.kill()   # Faster than 'stop'. Will also delete it so no need to be nice

在测试中等待整个服务器启动确实很烦人,但我想如果你想得更巧妙一点,可以缓存这个实例,在每次测试前清除它,或者在开发会话开始时启动一次,然后在下次会话开始时再刷新。我可能会考虑这样做。

如果你想要一个更长久的解决方案,不一定是程序化启动和关闭的,正常的Docker方式可能会更适合你。

4

我不知道有没有用Python实现的AMQP代理。而且我也不清楚有没有什么“轻量级”的实现;我觉得实现一个AMQP代理的过程相当复杂,所以那些尝试的人要么是想尽量接近AMQP规范的某个版本,要么就根本不想尝试。:)

我也不太明白,运行一个代理和为你的网页应用运行一个测试网页服务器有什么相同的问题。

网页服务器如果没有你的应用在里面,就没什么用。而在你开发应用的时候,能够在不进行完整部署的情况下运行它是很有意义的。

但是你并不是在实现代理的内部机制,而且你可以动态配置它,所以(和网页服务器不同)每次你修改代码时,它不需要自己重启。交换、绑定和队列可以由正在测试的应用声明,然后在之后自动删除。

安装RabbitMQ一点也不难,几乎不需要配置,因为它自带一个默认的虚拟主机和访客用户账户,这些在隔离的测试环境中使用是完全可以的。所以我从来没有遇到过在我的测试机器上简单运行RabbitMQ的问题。

也许你遇到了我没想到的特定问题;如果是这样,请留言(或者扩展你的问题)来解释一下。


编辑:最近我做了很多基于AMQP的应用测试,发现RabbitMQ的管理插件非常有用。它包括一个HTTP API,我用它来为每次测试运行创建一个新的虚拟主机,然后在测试结束后销毁它,以清理代理的状态。这使得在共享代理上运行测试变得不那么干扰。使用HTTP API来管理这些,而不是被测试的AMQP客户端,可以避免测试变得有些循环。

撰写回答