当前位置：首页 > news >正文

Python订单生成器+队列+异步提高性能和容错

news 2025/8/16 17:01:42

以下代码实现了一个订单生成器，使用 asyncio 和 aioredis 库实现了高并发地生成订单，并将新增订单异步更新到数据库。具体实现流程如下：

初始化 OrderGenerator 类。传入 Redis 服务器地址和并发数，在初始化函数中设置并发数和一个订单队列，用于存储新生成的订单数据。

定义 generate_order() 方法。该方法使用 Redis 的 WATCH、MULTI 和 EXEC 命令实现事务操作，生成一个订单数据，并将其添加到订单队列和 Redis 缓存中。如果生成订单的过程中监视到订单号已被修改，则重试生成订单，直到生成成功。

定义 update_database() 方法。该方法从订单队列中获取一个订单数据，模拟将其异步更新到数据库的操作，并将更新完成的订单标记为完成。

定义 generate_orders() 方法。该方法在循环中尝试连接 Redis 服务器，如果连接成功则创建多个 generate_order() 协程任务和一个 update_database() 协程任务，等待所有订单生成任务完成后等待订单队列中的所有任务完成，然后取消 update_database() 协程任务并关闭 Redis 连接。如果连接失败，则重试连接，最多重试三次。

import time
import random
import asyncio
import aioredisclass OrderGenerator:def __init__(self, redis_url, num_concurrent=10):# 初始化函数，传入 Redis 服务器地址和并发数self.redis_url = redis_urlself.num_concurrent = num_concurrent# 初始化订单队列，用于存储新生成的订单数据self.order_queue = asyncio.Queue()async def generate_order(self, redis):# 生成订单的协程任务async with redis.pipeline(transaction=True) as pipe:while True:try:# 使用 WATCH 命令监视订单号是否被修改await pipe.watch('order_id')# 获取当前订单号并自增order_id = await pipe.incr('order_id')# 使用 MULTI 命令开始一个事务pipe.multi()# 生成订单数据order_data = {'order_id': order_id,'user_id': random.randint(1, 100),'total_price': random.randint(100, 1000),'create_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}# 将订单添加到队列中await self.order_queue.put(order_data)# 将订单写入缓存await pipe.hmset(f'order:{order_id}', order_data)# 使用 EXEC 命令提交事务await pipe.execute()breakexcept aioredis.WatchError:# 如果订单号被修改，则重新尝试生成订单continueasync def update_database(self):# 将新生成的订单异步更新到数据库的协程任务while True:# 从订单队列中获取一个订单order_data = await self.order_queue.get()# 模拟将订单异步更新到数据库的操作await asyncio.sleep(1)print(f'订单 {order_data["order_id"]} 已更新到数据库')# 标记订单队列中的任务完成self.order_queue.task_done()async def generate_orders(self):retries = 0while retries < 3:# 与 Redis 服务器建立连接redis = await aioredis.Redis.from_url(self.redis_url)try:# 测试 Redis 服务器的可用性await redis.ping()# 创建并发数个生成订单的协程任务tasks = [asyncio.create_task(self.generate_order(redis)) for i in range(self.num_concurrent)]# 创建一个将新订单更新到数据库的协程任务update_task = asyncio.create_task(self.update_database())# 等待所有订单生成任务完成await asyncio.gather(*tasks)# 等待订单队列中的所有任务完成（即新订单更新到数据库）await self.order_queue.join()# 取消将新订单更新到数据库的协程任务update_task.cancel()# 关闭 Redis 连接redis.close()await redis.wait_closed()except (ConnectionError, TimeoutError):# 如果连接失败，则重试retries += 1print(f'Redis server is not available ({retries} retries)')redis.close()await redis.wait_closed()# 如果重试次数仍然失败，则抛出异常raise ConnectionError('Redis server is not available')if __name__ == '__main__':# 实例化 OrderGenerator 类，并运行生成订单的任务order_generator = OrderGenerator('redis://localhost', num_concurrent=10)asyncio.run(order_generator.generate_orders())