weenect/mock-amqp-server

mock-amqp-server

这是一个带监控功能的模拟AMQP服务器（兼容RabbitMQ），旨在通过网络层面测试消息发布者和消费者的交互行为，无需修改业务代码即可验证其正确性。

核心功能

• 模拟AMQP协议，支持队列、交换器的创建与绑定
• 通过HTTP接口实现消息的插入、查询、清除操作
• 监控消息的确认（ack）、拒绝（nack）及重新排队状态
• 支持二进制消息传输，自动处理base64编码
• 提供阻塞式API，便于集成测试中等待消息处理完成

使用场景

• CI/CD流水线中的集成测试，验证消息队列相关业务逻辑
• 排查生产环境中消息处理异常（如密码错误、回调顺序问题）
• 无需依赖真实RabbitMQ集群，快速搭建本地测试环境

配置说明

镜像默认暴露两个端口：

• 5672：AMQP服务端口，供业务服务连接
• 80：HTTP后门接口端口，用于测试控制

部署示例

Docker Run

bash
docker run -d \
  -p 5672:5672 \
  -p 8080:80 \
  --name mock-amqp \
  docker.xuanyuan.run/weenect/mock-amqp-server

Docker Compose

yaml
version: '3.8'
services:
  mock-amqp:
    image: docker.xuanyuan.run/weenect/mock-amqp-server
    ports:
      - "5672:5672"
      - "8080:80"
    restart: unless-stopped

可用HTTP接口

• DELETE /messages-in-queue/$QUEUE_NAME：清除指定队列中的消息
• DELETE /messages-in-exchange/$EXCHANGE_NAME：清除指定交换器中的消息
• GET /authentification-done-with-success-on/$USERNAME：检查指定用户是否成功连接
• GET /messages-acknowledged/$DELIVERY_TAG：阻塞等待消息被确认（ack），若已确认则直接返回
• GET /messages-not-acknowledged/$DELIVERY_TAG：阻塞等待消息被拒绝（nack），若已拒绝则直接返回
• GET /messages-requeued/$DELIVERY_TAG：阻塞等待消息被重新排队，若已重新排队则直接返回
• GET /messages-in-queue/$QUEUE_NAME：获取指定队列中待消费的消息（JSON格式，含base64标记）
• GET /messages-in-exchange/$EXCHANGE_NAME：获取指定交换器中待消费的消息（JSON格式，含base64标记）
• GET /queue-bound-to-exchange/$QUEUE_NAME/$EXCHANGE_NAME：阻塞等待队列与交换器绑定完成
• POST /add-message-on/$EXCHANGE_NAME：向指定交换器发送消息（支持JSON或二进制格式）
• POST /add-message-in-queue/$QUEUE_NAME：向指定队列发送消息（支持JSON或二进制格式）
• POST /create-exchange/$EXCHANGE_NAME：创建交换器
• POST /create-queue/$QUEUE_NAME：创建队列

消息发送细节

支持application/octet-stream类型传输二进制数据，消息头需以amqp_header_为前缀作为HTTP头传递。示例（Python）：

python
import requests
import pickle

message_body = {
    'action': 'my_action',
    'payload': {'id': 1, 'foo': 'bar'}
}
binary_body = pickle.dumps(message_body)

response = requests.post(
    "http://localhost:8080/add-message-on/my_exchange",
    headers={
        'Content-Type': 'application/octet-stream',
        'amqp_header_x-delay': '200'
    },
    data=binary_body
)
delivery_tag = int(response.content.decode('utf-8'))

特别感谢

感谢https://github.com/celery/py-amqp项目提供的类型序列化/反序列化代码。