脚本如何模拟消息生产

本文目录导读：

1. 基础脚本方式
2. 高级模拟方案
3. 分布式消息系统模拟
4. 性能测试工具
5. Docker 容器化部署
6. 最佳实践建议

我来详细说明脚本模拟消息生产的几种常见方法,从简单到复杂：

基础脚本方式

使用 Bash 脚本

#!/bin/bash
# 模拟消息生产到 Kafka
for i in {1..100}
do
  echo "{\"id\": $i, \"message\": \"test message $i\", \"timestamp\": \"$(date)\"}" | \
  kafka-console-producer.sh \
    --broker-list localhost:9092 \
    --topic test-topic
  sleep 1
done

使用 Python 脚本

import json
import time
from kafka import KafkaProducer
# 初始化生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
# 模拟消息生产
for i in range(100):
    message = {
        'id': i,
        'content': f'message_{i}',
        'timestamp': time.time()
    }
    producer.send('test-topic', message)
    print(f"Sent: {message}")
    time.sleep(1)
producer.close()

高级模拟方案

使用自定义数据生成器

import random
import string
from datetime import datetime
class MessageGenerator:
    def __init__(self):
        self.user_ids = ['user001', 'user002', 'user003']
        self.actions = ['click', 'view', 'purchase', 'login', 'logout']
        self.pages = ['home', 'product', 'cart', 'checkout', 'profile']
    def generate_message(self):
        return {
            'user_id': random.choice(self.user_ids),
            'action': random.choice(self.actions),
            'page': random.choice(self.pages),
            'timestamp': datetime.now().isoformat(),
            'session_id': ''.join(random.choices(string.ascii_letters, k=10)),
            'duration': random.randint(1, 300)
        }
# 使用
generator = MessageGenerator()
for _ in range(50):
    msg = generator.generate_message()
    print(json.dumps(msg))

分布式消息系统模拟

RabbitMQ 消息生产

import pika
import json
# 连接到 RabbitMQ
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()
# 声明队列
channel.queue_declare(queue='task_queue', durable=True)
# 生产消息
for i in range(100):
    message = json.dumps({
        'task_id': i,
        'type': random.choice(['email', 'sms', 'push']),
        'payload': f'task_data_{i}'
    })
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2,  # 持久化消息
        )
    )
    print(f" [x] Sent {message}")
connection.close()

Redis Pub/Sub

import redis
import json
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 模拟消息发布
for i in range(50):
    message = {
        'channel': 'notifications',
        'data': {
            'id': i,
            'type': random.choice(['info', 'warning', 'error']),
            'content': f'notification_{i}',
            'timestamp': time.time()
        }
    }
    r.publish('notifications', json.dumps(message))
    print(f"Published: {message}")
    time.sleep(0.5)

性能测试工具

Apache JMeter

<!-- JMeter 测试计划示例 -->
<TestPlan>
    <!-- 线程组配置 -->
    <ThreadGroup>
        <numThreads>10</numThreads>
        <rampTime>5</rampTime>
        <!-- 循环控制器 -->
        <LoopController>
            <loopCount>100</loopCount>
        </LoopController>
        <!-- Kafka 生产者采样器 -->
        <KafkaProducerSampler>
            <bootstrapServers>localhost:9092</bootstrapServers>
            <topicName>test-topic</topicName>
            <key>${__UUID()}</key>
            <value>{"test": "data"}</value>
        </KafkaProducerSampler>
    </ThreadGroup>
</TestPlan>

Docker 容器化部署

# docker-compose.yml
version: '3'
services:
  message-producer:
    image: python:3.9-slim
    volumes:
      - ./producer.py:/app/producer.py
    command: python /app/producer.py
    environment:
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
    depends_on:
      - kafka
  kafka:
    image: confluentinc/cp-kafka:latest
    environment:
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092

最佳实践建议

消息模板

class MessageTemplate:
    @staticmethod
    def create_order_message(order_id, user, items, total):
        return {
            "event_type": "order_created",
            "order_id": order_id,
            "user_id": user,
            "items": items,
            "total_amount": total,
            "timestamp": datetime.utcnow().isoformat(),
            "version": "1.0"
        }
    @staticmethod
    def create_user_action(action_type, user_id, details):
        return {
            "action_type": action_type,
            "user_id": user_id,
            "details": details,
            "timestamp": datetime.utcnow().isoformat()
        }

批量发送与重试

def batch_send(producer, topic, messages, max_retries=3):
    for msg in messages:
        for attempt in range(max_retries):
            try:
                future = producer.send(topic, msg)
                result = future.get(timeout=10)
                break
            except Exception as e:
                print(f"Attempt {attempt + 1} failed: {e}")
                time.sleep(2 ** attempt)  # 指数退避

选择合适的模拟方案取决于：

• 数据格式：JSON、Avro、Protobuf
• 消息系统：Kafka、RabbitMQ、Redis、ActiveMQ
• 性能要求：TPS、延迟要求
• 复杂度：是否包含业务逻辑

建议从小规模开始,逐步增加复杂度和负载，同时做好监控和日志记录。