实例解析：秒杀系统是如何防止商品超售

📝 面试求职：  ，内容涵盖 测试基础、Linux操作系统、MySQL数据库、Web功能测试、接口测试、APPium移动端测试、Python知识、Selenium自动化测试相关、性能测试、性能测试、计算机网络知识、Jmeter、HR面试，命中率杠杠的。（大家刷起来…）

📝 职场经验干货：

秒杀系统的基本思路

秒杀系统的主要目标是在高并发情况下，确保用户能够安全地购买商品，同时防止超卖（即库存不足时仍然允许用户下单）。为了实现这一目标，我们需要考虑以下几个方面：

实现原理

1. 线程安全的库存管理：

   • 我们使用 queue.Queue 来管理库存。这个队列的大小由 total_stock 决定，表示可用库存的数量。

   • 每个库存项用 1 表示，初始化时将 total_stock 个 1 放入队列中。这样，队列的长度始终代表当前可用的库存。

2. 线程同步：

   • 使用 threading.Lock() 创建一个锁对象 self.lock。在处理秒杀请求时，我们需要确保对库存的操作是原子性的，即在检查库存和扣减库存时，其他线程不能同时访问。

   • 通过 with self.lock: 语句来获取和释放锁，确保在这个代码块内的操作不会被其他线程打断。

3. 秒杀逻辑：

   • 在 seckill 方法中，首先尝试获取锁。如果成功获取锁，接下来检查库存是否可用。

   • 使用 self.stock_queue.empty() 检查库存。如果库存不为空，调用 self.stock_queue.get_nowait() 从队列中取出一个库存项，表示成功售出一个商品。

   • 如果成功售出商品，调用 self.create_order(user_id) 模拟订单的创建，并更新成功计数器 self.success_count。如果库存不足，则更新失败计数器 self.fail_count。

4. 异常处理：

   • 在 seckill 方法中，使用 try-except 结构来捕获可能的异常，确保系统的稳定性。如果库存队列为空，捕获 queue.Empty 异常并更新失败计数。

5. 并发模拟：

   • 使用 ThreadPoolExecutor 来模拟500个用户同时请求秒杀。通过设置 max_workers=200，我们可以同时处理200个线程。

   • executor.submit(seckill_system.seckill, user_id) 将每个用户的秒杀请求提交给线程池进行处理。

6. 结果统计：

   • 在所有线程完成后，打印总库存、成功秒杀数和失败秒杀数，以便分析秒杀的结果。

代码示例：

import threading  
import queue  
import time  
from concurrent.futures import ThreadPoolExecutor  

class ProductSecKill:
def __init__(self, total_stock):
"""  
        初始化秒杀系统  
        :param total_stock: 商品总库存  
        """
# 使用线程安全的队列管理库存  
        self.stock_queue = queue.Queue(total_stock)  

# 初始化库存  
for _ in range(total_stock):  
            self.stock_queue.put(1)  

# 线程同步锁  
        self.lock = threading.Lock()  

# 成功秒杀计数器  
        self.success_count = 0

# 失败秒杀计数器  
        self.fail_count = 0

def seckill(self, user_id):
"""  
        秒杀业务逻辑  
        :param user_id: 用户ID  
        :return: 秒杀结果  
        """
try:  
# 使用锁确保原子操作  
with self.lock:  
# 检查是否还有库存  
if not self.stock_queue.empty():  
# 从队列中获取一个库存项  
                    self.stock_queue.get_nowait()  

# 模拟订单创建  
                    self.create_order(user_id)  

# 成功秒杀计数  
                    self.success_count += 1
return True
else:  
# 库存不足  
                    self.fail_count += 1
return False
except queue.Empty:  
# 库存已售罄  
            self.fail_count += 1
return False
except Exception as e:  
            print(f"秒杀异常: {e}")  
return False

def create_order(self, user_id):
"""  
        创建订单（模拟）  
        :param user_id: 用户ID  
        """
        print(f"用户 {user_id} 秒杀成功，创建订单")  

def simulate_seckill():
"""  
    模拟秒杀场景  
    """
# 总库存设置为100  
    total_stock = 100

# 创建秒杀实例  
    seckill_system = ProductSecKill(total_stock)  

# 使用线程池模拟并发  
with ThreadPoolExecutor(max_workers=200) as executor:  
# 模拟500个用户同时秒杀  
        futures = [  
            executor.submit(seckill_system.seckill, user_id)   
for user_id in range(500)  
        ]  

# 等待所有线程完成  
for future in futures:  
            future.result()  

# 打印结果  
    print(f"总库存: {total_stock}")  
    print(f"成功秒杀数: {seckill_system.success_count}")  
    print(f"失败秒杀数: {seckill_system.fail_count}")  

# 运行秒杀模拟  
if __name__ == "__main__":  
    simulate_seckill()

代码流程

• 初始化：创建 ProductSecKill 实例，初始化库存。

• 秒杀请求：每个用户请求秒杀时，调用 seckill 方法。

• 库存检查与扣减：在获取锁后，检查库存并处理订单。

• 统计结果：所有请求处理完毕后，输出成功和失败的统计信息。

优化建议：

1. 实际生产中可以引入分布式锁（如Redis）

2. 增加更复杂的库存和订单管理逻辑

3. 添加更详细的异常处理和日志记录

总结：

这个秒杀系统通过使用线程安全的队列和锁机制，有效地管理了高并发情况下的库存，确保了每个用户的请求都能得到合理的处理。通过模拟多个用户的并发请求，我们可以测试系统在实际使用中的表现，确保不会出现超卖的情况。这种设计适合于电商平台的秒杀活动，能够在高负载下保持稳定。

最后： 下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取【保证100%免费】