1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 拉取RabbitMQ镜像（带管理界面；版本号请取当前稳定版，4.x 已是主流）
docker pull rabbitmq:4-management

# 运行RabbitMQ容器
docker run -d \
    --name rabbitmq \
    -p 5672:5672 \
    -p 15672:15672 \
    -e RABBITMQ_DEFAULT_USER=admin \
    -e RABBITMQ_DEFAULT_PASS=Admin123 \
    rabbitmq:4-management

# 查看运行状态
docker ps

# 查看日志
docker logs rabbitmq

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 方式A：用发行版自带的包（最简单，版本通常略旧）
sudo apt update
sudo apt install -y rabbitmq-server

# 方式B：用 RabbitMQ 官方仓库装新版本（推荐生产环境）
# 1) 准备工具
sudo apt install -y curl gnupg apt-transport-https
# 2) 导入官方签名密钥
curl -1sLf "https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/gpg.key" \
  | sudo gpg --dearmor -o /usr/share/keyrings/rabbitmq.gpg
# 3) 添加软件源——注意把 "noble" 换成你系统的代号（lsb_release -cs 可以查到）
sudo tee /etc/apt/sources.list.d/rabbitmq.list >/dev/null <<'EOF'
deb [signed-by=/usr/share/keyrings/rabbitmq.gpg] https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/deb/ubuntu noble main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.gpg] https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/deb/ubuntu noble main
EOF
# 4) 安装
sudo apt update
sudo apt install -y rabbitmq-server

# 启动服务
sudo systemctl enable --now rabbitmq-server

# 启用自带的管理插件（Web界面）
sudo rabbitmq-plugins enable rabbitmq_management

# 查看状态
sudo systemctl status rabbitmq-server

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# rabbitmqadmin 并不在容器里，它是"管理插件"提供的一个 Python 脚本，
# 需要先从管理界面下载（前提：已启用 rabbitmq_management 插件）
curl -O http://localhost:15672/cli/rabbitmqadmin
chmod +x rabbitmqadmin

# 用它创建交换机
./rabbitmqadmin -u admin -p Admin123 declare exchange \
    name=my-exchange \
    type=direct \
    durable=true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 声明队列（用上一步下载好的 rabbitmqadmin）
./rabbitmqadmin -u admin -p Admin123 declare queue \
    name=my-queue \
    durable=true

# 列出所有队列
./rabbitmqadmin -u admin -p Admin123 list queues

# 绑定队列到交换机
./rabbitmqadmin -u admin -p Admin123 declare binding \
    source=my-exchange \
    destination=my-queue \
    routing_key=my-key

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 安装pika库
pip install pika

# producer.py
import pika

# 连接RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# 声明交换机
channel.exchange_declare(
    exchange='my-exchange',
    exchange_type='direct',
    durable=True
)

# 发送消息
message = "Hello RabbitMQ!"
channel.basic_publish(
    exchange='my-exchange',
    routing_key='my-key',  # 路由键
    body=message,
    properties=pika.BasicProperties(
        delivery_mode=2,  # 持久化消息
        content_type='text/plain'
    )
)

print(f" [x] Sent: {message}")

connection.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# consumer.py
import pika

# 连接RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# 声明队列（和发送者保持一致）
channel.queue_declare(queue='my-queue', durable=True)

# 定义回调函数
def callback(ch, method, properties, body):
    print(f" [x] Received: {body.decode()}")
    
    # 模拟处理
    import time
    time.sleep(1)
    
    # 确认消息已处理
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 设置QoS（预取数量）
channel.basic_qos(prefetch_count=1)

# 开始消费
channel.basic_consume(
    queue='my-queue',
    on_message_callback=callback,
    auto_ack=False  # 手动确认
)

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# producer.py - 生产者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明交换机
channel.exchange_declare(exchange='orders', exchange_type='direct', durable=True)

# 发送订单消息
order = '{"order_id": 12345, "amount": 99.99, "user": "xiaoming"}'
channel.basic_publish(
    exchange='orders',
    routing_key='new_order',
    body=order,
    properties=pika.BasicProperties(
        delivery_mode=2,
        content_type='application/json'
    )
)

print(f" [x] Order sent: {order}")
connection.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# consumer.py - 消费者
import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明交换机和队列
channel.exchange_declare(exchange='orders', exchange_type='direct', durable=True)
channel.queue_declare(queue='order_processing', durable=True)

# 绑定队列到交换机
channel.queue_bind(exchange='orders', queue='order_processing', routing_key='new_order')

def process_order(ch, method, properties, body):
    order = json.loads(body)
    print(f" [x] Processing order: {order}")
    
    # 模拟订单处理（库存扣减、支付等）
    print(f"     Order {order['order_id']} processed!")
    
    # 确认消息
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='order_processing', on_message_callback=process_order)

print(' [*] Waiting for orders...')
channel.start_consuming()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# 设置完整环境的脚本
import pika

def setup_rabbitmq():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # 1. 创建交换机
    channel.exchange_declare(
        exchange='orders',
        exchange_type='direct',
        durable=True
    )
    
    # 2. 创建队列
    channel.queue_declare(queue='order_processing', durable=True)
    channel.queue_declare(queue='order_notification', durable=True)
    
    # 3. 绑定队列到交换机
    channel.queue_bind(
        exchange='orders',
        queue='order_processing',
        routing_key='new_order'
    )
    channel.queue_bind(
        exchange='orders',
        queue='order_notification',
        routing_key='new_order'
    )
    
    print(" [*] RabbitMQ setup complete!")
    connection.close()

if __name__ == '__main__':
    setup_rabbitmq()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 用 Docker Compose 起一个单节点 Kafka（KRaft 模式，不需要 ZooKeeper）
# 文件名：compose.yaml
cat > compose.yaml <<'EOF'
services:
  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
EOF

docker compose up -d

# 查看Kafka日志
docker logs -f kafka

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 用 Kafka-UI 提供 Web 管理界面
# 注意：老教程里的 hlebalbau/kafka-manager 早已停止维护（且依赖 ZooKeeper），
#       现在更常用 provectuslabs/kafka-ui 或 AKHQ
docker run -d \
    --name kafka-ui \
    -p 8080:8080 \
    -e KAFKA_CLUSTERS_0_NAME=local \
    -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=localhost:9092 \
    --network container:kafka \
    provectuslabs/kafka-ui:latest

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 进入Kafka容器
docker exec -it kafka bash

# 创建Topic
kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --topic my-topic \
    --partitions 3 \
    --replication-factor 1

# 查看Topic列表
kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看Topic详情
kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-topic

1
2
3
4
5
6
7

# 默认策略：根据key的hash值分配分区
# 有key的消息，相同key一定到同一个分区
producer.send('my-topic', key='user:1', value='message')

# 无key的消息：轮询分配
producer.send('my-topic', value='message1')
producer.send('my-topic', value='message2')  # 会分到下一个分区

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# ⚠️ 消费者组**不需要也不能手工"创建"**：只要一个消费者用 group.id=my-group
#    开始订阅 Topic，这个组就自动出现了（组是消费时"浮现"出来的）。
#    所以 kafka-consumer-groups.sh 里没有 --create 这个选项，老教程写错了。

# 查看消费者组列表
kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# 查看消费者组详情
kafka-consumer-groups.sh --describe \
    --bootstrap-server localhost:9092 \
    --group my-group

# 常用运维操作：重置位移（注意：必须先把消费者停掉）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group my-group --topic my-topic \
    --reset-offsets --to-earliest --execute

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 安装kafka库
# 说明：kafka-python 用纯 Python 实现，上手简单、依赖少，适合学习和轻量场景；
#       生产上更常用官方维护、基于 librdkafka 的 confluent-kafka（性能更好）：
#       pip install confluent-kafka
pip install kafka-python

# producer.py
from kafka import KafkaProducer

# 创建生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: v.encode('utf-8')
)

# 发送消息
for i in range(10):
    message = f"Message {i}"
    producer.send('my-topic', value=message)
    print(f" [x] Sent: {message}")

# 确保所有消息发送完成
producer.flush()
producer.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# consumer.py
from kafka import KafkaConsumer

# 创建消费者
consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',  # 消费者组
    auto_offset_reset='earliest',  # 从最早的消息开始消费
    value_deserializer=lambda v: v.decode('utf-8')
)

print(' [*] Waiting for messages...')

for message in consumer:
    print(f" [x] Received: {message.value}")
    print(f"      Partition: {message.partition}, Offset: {message.offset}")