Jmeter混合场景压测

from flask import Flask, jsonify, request
import time
import random

app = Flask(__name__)

# 模拟数据
flights = [
    {"id": 1, "flight_no": "CA123", "from": "PEK", "to": "SHA", "price": 980, "seats": 150},
    {"id": 2, "flight_no": "MU456", "from": "SHA", "to": "CAN", "price": 750, "seats": 180},
    {"id": 3, "flight_no": "CZ789", "from": "CAN", "to": "CTU", "price": 650, "seats": 120},
]

bookings = {}
booking_counter = 1

@app.route('/search', methods=['GET'])
def  search_flights():
    """航班搜索接口"""
    # 模拟处理时间 (50-150ms)
    time.sleep(random.uniform(0.05, 0.15))
    
    from_city = request.args.get('from')
    to_city = request.args.get('to')
    date = request.args.get('date')
    
    # 简单过滤逻辑
    results = [f for f in flights if f['from'] == from_city and f['to'] == to_city]
    
    return jsonify({
        "status": "success",
        "results": results,
        "count": len(results)
    })

@app.route('/book', methods=['POST'])
def  book_flight():
    """航班预订接口"""
    # 模拟处理时间 (100-300ms)
    time.sleep(random.uniform(0.1, 0.3))
    
    data = request.json
    flight_id = data.get('flight_id')
    passenger = data.get('passenger')
    
    # 查找航班
    flight = next((f for f in flights if f['id'] == flight_id), None)
    
    if  not flight:
        return jsonify({"status": "error", "message": "Flight not found"}), 404
    
    if flight['seats'] <= 0:
        return jsonify({"status": "error", "message": "No seats available"}), 400
    
    # 创建预订
    global booking_counter
    booking_id = booking_counter
    booking_counter += 1
    
    bookings[booking_id] = {
        "booking_id": booking_id,
        "flight": flight,
        "passenger": passenger,
        "status": "confirmed"
    }
    
    # 减少座位
    flight['seats'] -= 1
    
    return jsonify({
        "status": "success",
        "booking_id": booking_id,
        "flight": flight,
        "passenger": passenger
    })

@app.route('/pay', methods=['POST'])
def  make_payment():
    """支付接口"""
    # 模拟处理时间 (200-500ms)
    time.sleep(random.uniform(0.2, 0.5))
    
    data = request.json
    booking_id = data.get('booking_id')
    payment_info = data.get('payment_info')
    
    # 查找预订
    booking = bookings.get(booking_id)
    
    if  not booking:
        return jsonify({"status": "error", "message": "Booking not found"}), 404
    
    # 模拟支付处理
    if random.random() < 0.95:  # 95%成功率
        booking['status'] = "paid"
        return jsonify({
            "status": "success",
            "payment_id": random.randint(100000, 999999),
            "booking": booking
        })
    else:
        return jsonify({
            "status": "error",
            "message": "Payment failed"
        }), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, threaded=True)

pip install flask
python flight_api.py

{
  "flight_id": ${flight_id},
  "passenger": {
    "name": "Test User",
    "id_card": "110101199001011234"
  }
}

// 随机选择一个航班ID
def flights = [1, 2, 3] // 实际应用中应从搜索响应中获取
vars.put("flight_id", flights.get(new Random().nextInt(flights.size()) as String)

{
  "booking_id": ${booking_id},
  "payment_info": {
    "card": "4111111111111111",
    "expiry": "12/25"
  }
}

// 从之前的预订中获取booking_id
if (vars.get("last_booking_id") != null) {
    vars.put("booking_id", vars.get("last_booking_id"))
} else {
    // 如果没有预订记录，使用默认值
    vars.put("booking_id", "1")
}