4. 环境搭建与基础实践:Docker Compose 搭建 RabbitMQ + Kafka 集群,生产与消费第一条消息

好,咱们进入实战环节。说实话,很多人在消息队列上栽跟头,不是因为原理不懂,而是环境没搭对。我见过太多人花一整天在本地装 RabbitMQ,结果依赖冲突、端口被占、Erlang 版本不对……折腾到怀疑人生。

所以这一章,我带你用 Docker Compose 一把梭。你只需要一个 Docker 环境,十分钟内就能跑起两个完整的消息队列集群。

4.1 为什么选 Docker Compose?

我个人习惯,所有中间件的本地开发环境都用 Docker Compose 管理。原因很简单:

  • 环境隔离——不会污染你的宿主机
  • 版本可控——镜像标签锁死,团队一致
  • 一键启停——docker-compose up -d 搞定
  • 集群模拟——单机就能跑多节点
小提示:生产环境当然不会用 Docker Compose,但本地开发和调试,它是最快的路径。你想想看,要是每次调试都要申请一套物理机,黄花菜都凉了。

4.2 环境准备

先确认你的机器上有这些:

组件 最低版本 检查命令
Docker 20.10+ docker --version
Docker Compose v2.0+ docker compose version
Git 任意 git --version

嗯,这里要注意:Docker Compose v1 和 v2 的命令格式不一样。v2 是 docker compose(中间没横杠),我下面的示例全用 v2 语法。

4.3 项目结构

我习惯把每个中间件的配置分开,方便管理。创建一个工作目录:

mkdir -p ~/mq-workshop/chapter4
cd ~/mq-workshop/chapter4

目录结构长这样:

chapter4/
├── rabbitmq/
│   ├── docker-compose.yml
│   └── rabbitmq.conf
└── kafka/
    ├── docker-compose.yml
    └── server.properties

4.4 搭建 RabbitMQ 集群

先搞 RabbitMQ。我个人比较喜欢用 3.12 版本,稳定且功能全。创建一个 rabbitmq/docker-compose.yml:

version: '3.8'

services:
  rabbitmq:
    image: rabbitmq:3.12-management
    container_name: rabbitmq-node1
    hostname: rabbitmq-node1
    ports:
      - "5672:5672"    # AMQP 协议端口
      - "15672:15672"  # 管理界面端口
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123
      RABBITMQ_ERLANG_COOKIE: "SWQOKODSQALRPCLNMEQG"
    volumes:
      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
    networks:
      - mq-network

networks:
  mq-network:
    driver: bridge

再创建一个 rabbitmq/rabbitmq.conf:

loopback_users.guest = false
listeners.tcp.default = 5672
management.tcp.port = 15672
disk_free_limit.absolute = 2GB
避坑指南:我曾经在生产环境忘了设置 disk_free_limit,结果磁盘快满时 RabbitMQ 直接阻塞所有生产者,线上报警炸了。这个参数一定要配,至少留 1-2GB 的余量。

启动:

cd rabbitmq
docker compose up -d

等几秒,访问 http://localhost:15672,用 admin / admin123 登录。看到管理界面,说明 RabbitMQ 跑起来了。

4.5 搭建 Kafka 集群

Kafka 比 RabbitMQ 稍微复杂一点,因为它依赖 ZooKeeper。不过 Docker Compose 可以一起搞定。

创建 kafka/docker-compose.yml:

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
    networks:
      - kafka-network

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka-broker1
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    volumes:
      - ./server.properties:/etc/kafka/server.properties
    networks:
      - kafka-network

networks:
  kafka-network:
    driver: bridge

这里有个关键点:KAFKA_ADVERTISED_LISTENERS 必须配成 localhost:9092,否则客户端连不上。我在项目里踩过这个坑,容器内和容器外的网络地址不一样,不配这个参数,客户端永远连不上。

启动:

cd kafka
docker compose up -d

验证一下:

docker logs kafka-broker1 | grep "started"

看到 [KafkaServer id=1] started,说明 Kafka 起来了。

4.6 知识结构总览

下面这张图,帮你理清两个消息队列的架构差异:

RabbitMQ vs Kafka 架构对比 RabbitMQ 生产者 消费者 交换机 (Exchange) 队列 (Queue) Kafka 生产者 消费者 Topic (主题) Partition (分区) 核心:Exchange → Binding → Queue 消息被消费后删除(默认) 核心:Producer → Topic → Partition 消息持久化,按 offset 消费

4.7 生产与消费第一条消息

环境跑起来了,咱们来发第一条消息。先搞 RabbitMQ。

4.7.1 RabbitMQ 第一条消息

用 Python 写个简单的生产者和消费者。先装依赖:

pip install pika

生产者 (producer_rabbit.py):

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost', 5672, '/',
        credentials=pika.PlainCredentials('admin', 'admin123'))
)
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_publish(exchange='', routing_key='hello', body='Hello, RabbitMQ!')
print(" [x] Sent 'Hello, RabbitMQ!'")

connection.close()

消费者 (consumer_rabbit.py):

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost', 5672, '/',
        credentials=pika.PlainCredentials('admin', 'admin123'))
)
channel = connection.channel()

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print(f" [x] Received {body}")

channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

先开一个终端跑消费者,再开另一个终端跑生产者。你会看到消费者打印出 [x] Received Hello, RabbitMQ!。第一条消息,搞定。

4.7.2 Kafka 第一条消息

Kafka 官方提供了命令行工具,咱们直接用。先进入容器:

docker exec -it kafka-broker1 bash

创建一个 Topic:

kafka-topics --create \
  --topic hello-topic \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1

启动生产者,发送消息:

kafka-console-producer --topic hello-topic --bootstrap-server localhost:9092

输入 Hello, Kafka!,然后按 Ctrl+C 退出。

再开一个终端,启动消费者:

docker exec -it kafka-broker1 bash
kafka-console-consumer --topic hello-topic --bootstrap-server localhost:9092 --from-beginning

你会看到 Hello, Kafka! 打印出来。注意 --from-beginning 参数,它会从头消费所有消息。不加的话,只消费新消息。

关键区别:RabbitMQ 的消息被消费后默认删除,而 Kafka 的消息会持久化在磁盘上,按 offset 保留。这就是为什么 Kafka 的消费者可以 --from-beginning 重新消费历史消息。

4.8 验证集群状态

跑完基础流程,咱们验证一下集群是否健康。

组件 验证方式 预期结果
RabbitMQ 浏览器访问 localhost:15672 管理界面正常,队列 hello 可见
Kafka kafka-topics --list --bootstrap-server localhost:9092 输出 hello-topic
Kafka kafka-consumer-groups --bootstrap-server localhost:9092 --list 输出消费者组列表

4.9 常见问题与避坑

  • 端口冲突:如果 5672 或 9092 被占用,改一下宿主机映射端口。比如 5673:5672,客户端连接时也要改端口。
  • Kafka 连不上:八成是 KAFKA_ADVERTISED_LISTENERS 没配对。容器内用容器名,容器外用 localhost。
  • RabbitMQ 管理界面打不开:确认镜像带 management 标签,不带 management 的镜像没有 Web 界面。
  • Python 客户端报错:检查 pika 版本,0.13+ 的 API 有变化。我用的 1.3.2,上面的代码没问题。
重要提醒:这些环境只适合本地开发和调试。生产环境一定要考虑高可用、持久化、监控告警。别把 docker-compose 直接搬到生产上,我见过有人这么干,结果容器挂了消息全丢。

好,环境搭好了,第一条消息也发出去了。接下来你可以自由探索,比如试试 RabbitMQ 的 Topic 交换机,或者 Kafka 的多分区消费。动手才是最好的学习方式。