4. 消息队列核心机制

好，咱们进入实战环节。说实话，很多人在消息队列上栽跟头，不是因为原理不懂，而是环境没搭对。我见过太多人花一整天在本地装 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 的多分区消费。动手才是最好的学习方式。

← 上一章 📖 返回目录 下一章 →