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

# 拉取RabbitMQ镜像（带管理界面）
docker pull rabbitmq:3.12-management

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

# 查看运行状态
docker ps

# 查看日志
docker logs rabbitmq

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

# 安装Erlang（RabbitMQ依赖Erlang）
sudo apt update
sudo apt install -y erlang

# 添加RabbitMQ仓库
curl -fsSL https://packagecloud.io/rabbitmq/rabbitmq-server/gpgkey | sudo gpg --dearmor -o /usr/share/keyrings/rabbitmq.gpg

# 安装RabbitMQ
sudo apt update
sudo apt install -y rabbitmq-server

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

# 查看状态
sudo systemctl status rabbitmq-server

1
2
3
4
5

# 使用rabbitmqadmin创建交换机
docker exec rabbitmq rabbitmqadmin declare exchange \
    name=my-exchange \
    type=direct \
    durable=true

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

# 声明队列
docker exec rabbitmq rabbitmqadmin declare queue \
    name=my-queue \
    durable=true

# 列出所有队列
docker exec rabbitmq rabbitmqadmin list queues

# 绑定队列到交换机
docker exec rabbitmq rabbitmqadmin 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

# 创建一个网络（可选）
docker network create kafka-net

# 启动Zookeeper（Kafka依赖Zookeeper）
docker run -d \
    --name zookeeper \
    -p 2181:2181 \
    confluentinc/cp-zookeeper:latest \
    environment ZOOKEEPER_CLIENT_PORT=2181

# 启动Kafka
docker run -d \
    --name kafka \
    --network kafka-net \
    -p 9092:9092 \
    -e KAFKA_BROKER_ID=1 \
    -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
    -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
    confluentinc/cp-kafka:latest

# 查看Kafka日志
docker logs -f kafka

1
2
3
4
5
6

# 安装kafka-manager（一个Web管理界面）
docker run -d \
    --name kafka-manager \
    -p 9000:9000 \
    -e ZK_HOSTS=zookeeper:2181 \
    hlebalbau/kafka-manager:stable

 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

# 创建消费者组
kafka-consumer-groups.sh --create \
    --bootstrap-server localhost:9092 \
    --group my-group \
    --topic my-topic

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

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

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

# 安装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}")