从0到1搭建实时通信测试：一个订单流失1.2亿，只因延迟慢了200毫秒_千锋教育

import time
import threading

def order_matching(order):
    start_time = time.perf_counter_ns()
    
    # 订单校验
    validate_order(order)
    
    # 资金冻结
    freeze_balance(order)
    
    # 撮合匹配
    match_result = match_engine(order)
    
    end_time = time.perf_counter_ns()
    latency = (end_time - start_time) / 1_000_000  # 转换为毫秒
    
    if latency > 200:
        alert_team(f"订单{order.id}延迟{latency}ms，超出阈值")
    
    return match_result

import asyncio
import aiohttp
import random

async def send_order(session, order_type):
    order = {
        "symbol": "BTC/USDT",
        "side": random.choice(["buy", "sell"]),
        "type": order_type,
        "price": random.uniform(50000, 60000) if order_type == "limit" else None,
        "quantity": random.uniform(0.01, 1.0)
    }
    
    async with session.post("https://api.exchange.com/orders", json=order) as resp:
        return await resp.json()

async def concurrency_test(concurrent_orders):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(concurrent_orders):
            order_type = random.choices(
                ["market", "limit", "cancel"],
                weights=[0.7, 0.2, 0.1]
            )[0]
            tasks.append(send_order(session, order_type))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        success_count = sum(1 for r in results if isinstance(r, dict) and r.get("status") == "ok")
        failed_count = sum(1 for r in results if isinstance(r, Exception))
        
        print(f"成功: {success_count}, 失败: {failed_count}")

# 执行10万并发测试
asyncio.run(concurrency_test(100000))

import redis
import mysql.connector

def check_data_consistency(order_id):
    # 连接Redis缓存
    r = redis.Redis(host='localhost', port=6379, db=0)
    cached_order = r.get(f"order:{order_id}")
    
    # 连接MySQL数据库
    conn = mysql.connector.connect(
        host='localhost',
        database='exchange',
        user='test',
        password='test'
    )
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM orders WHERE id = %s", (order_id,))
    db_order = cursor.fetchone()
    
    if cached_order and db_order:
        # 比较关键字段
        if cached_order['status'] != db_order[3]:
            print(f"数据不一致：订单{order_id}状态缓存={cached_order['status']}, 数据库={db_order[3]}")
            return False
    return True

# 每5分钟执行一次校验
import time
while True:
    order_id = random.randint(1, 1000000)
    if not check_data_consistency(order_id):
        alert_team(f"数据一致性检查失败，订单ID: {order_id}")
    time.sleep(300)

import hashlib
import hmac
import time
import requests

def generate_signature(secret, params):
    # 按字母排序参数
    sorted_params = sorted(params.items())
    query_string = '&'.join([f"{k}={v}" for k, v in sorted_params])
    
    # 生成签名
    signature = hmac.new(
        secret.encode(),
        query_string.encode(),
        hashlib.sha256
    ).hexdigest()
    
    return signature

def test_signature_bypass():
    # 尝试绕过签名
    url = "https://api.exchange.com/withdraw"
    
    # 场景1：空签名
    params = {"amount": "1", "address": "test"}
    resp = requests.post(url, params=params)
    assert resp.status_code == 401, f"空签名返回{resp.status_code}"
    
    # 场景2：过期签名
    old_timestamp = int(time.time()) - 3600  # 1小时前
    params["timestamp"] = old_timestamp
    params["signature"] = generate_signature("secret", params)
    resp = requests.post(url, params=params)
    assert resp.status_code == 401, f"过期签名返回{resp.status_code}"
    
    # 场景3：重放攻击
    params["timestamp"] = int(time.time())
    params["signature"] = generate_signature("secret", params)
    resp1 = requests.post(url, params=params)
    resp2 = requests.post(url, params=params)  # 重放
    assert resp2.status_code == 401, f"重放攻击返回{resp2.status_code}"
    
    print("所有签名绕过测试通过")

import random
import string

def test_kyc_bypass():
    # 尝试绕过KYC
    test_cases = [
        {"name": "Test User", "id_number": "123456789", "photo": "fake_face.jpg"},
        {"name": "", "id_number": "000000000", "photo": "blank.jpg"},
        {"name": "Admin", "id_number": "admin123", "photo": "system_photo.jpg"}
    ]
    
    for case in test_cases:
        resp = kyc_verification(case)
        if resp["status"] == "approved":
            print(f"KYC绕过成功：{case['name']}")
            return False
    
    print("所有KYC绕过测试通过")
    return True