1. 实时交易数据模拟器

任务说明

构建一个可配置的 Python 脚本，持续生成模拟交易流。每条交易包含：用户ID、金额、时间戳、商户类别。输出目标支持两种模式：本地内存队列（用于单机测试）或 Kafka（用于后续反欺诈模块联调）。

环境与依赖

• Python 3.8+
• 本地队列模式：仅用标准库 queue，无需额外依赖
• Kafka 模式：pip install kafka-python，并确保 Kafka 可访问

核心实现

脚本通过命令行参数配置：生成速率、总条数、输出模式、Kafka 地址与主题。交易字段随机生成，时间戳为毫秒级。本地模式把交易放入 queue.Queue，供同进程消费者读取；Kafka 模式逐条发送到指定主题。

import argparse
import json
import queue
import random
import time
from datetime import datetime

MERCHANTS = ["grocery", "electronics", "travel", "gaming", "fuel", "fashion"]
USER_IDS = [f"u{1000 + i}" for i in range(50)]

def make_txn():
    return {
        "user_id": random.choice(USER_IDS),
        "amount": round(random.uniform(1.0, 5000.0), 2),
        "timestamp": int(datetime.now().timestamp() * 1000),
        "merchant_category": random.choice(MERCHANTS),
    }

def run_local(count, rate, out_queue):
    for _ in range(count):
        out_queue.put(make_txn())
        time.sleep(1.0 / rate)
    out_queue.put(None)  # 结束标记

def run_kafka(count, rate, bootstrap, topic):
    from kafka import KafkaProducer
    producer = KafkaProducer(
        bootstrap_servers=bootstrap,
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    )
    for _ in range(count):
        producer.send(topic, make_txn())
        time.sleep(1.0 / rate)
    producer.flush()
    producer.close()

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--mode", choices=["local", "kafka"], default="local")
    p.add_argument("--count", type=int, default=20)
    p.add_argument("--rate", type=float, default=5.0, help="每秒生成条数")
    p.add_argument("--bootstrap", default="localhost:9092")
    p.add_argument("--topic", default="transactions")
    args = p.parse_args()

    if args.mode == "local":
        q = queue.Queue()
        run_local(args.count, args.rate, q)
        while True:
            txn = q.get()
            if txn is None:
                break
            print(json.dumps(txn, ensure_ascii=False))
    else:
        run_kafka(args.count, args.rate, args.bootstrap, args.topic)
        print(f"sent {args.count} txns to {args.topic}")

if __name__ == "__main__":
    main()

运行方式

# 本地队列模式，生成 20 条，每秒 5 条
python simulator.py --mode local --count 20 --rate 5

# Kafka 模式
python simulator.py --mode kafka --count 100 --rate 10 \
  --bootstrap localhost:9092 --topic transactions

验证方法

• 本地模式：终端应逐行打印 JSON，每行含 user_id、amount、timestamp、merchant_category 四个字段，条数等于 --count，输出节奏接近 --rate。
• Kafka 模式：命令结束后打印 sent N txns；用 kafka-console-consumer --topic transactions --from-beginning 可看到相同结构的 JSON 消息。
• 通过条件：字段齐全、JSON 可解析、条数与参数一致。

接入后续反欺诈模块

本地模式下，把 queue.Queue 实例传给反欺诈消费者即可在同一进程内消费；Kafka 模式下，反欺诈模块订阅同一主题即可。两种模式产出的消息结构完全一致，切换时无需改动下游解析逻辑。