MongoDB Test5: 测试并发性能

需求

代码

import asyncio
import pymongo
import random
import time

async def insert_document(collection, user_id, value):
    await collection.insert_one({'user_id': user_id, 'value': value})

async def test_scenario(loop, client, user_id):
    db = client['test_db']
    collection = db['test_collection']

    for _ in range(100):  # 每个用户执行100次操作
        value = random.randint(1, 1000)
        await loop.run_in_executor(None, insert_document, collection, user_id, value)

async def main(num_users):
    client = pymongo.MongoClient("192.168.92.159", 20000, maxPoolSize=num_users)

    loop = asyncio.get_running_loop()

    tasks = []
    for user_id in range(num_users):
        tasks.append(test_scenario(loop, client, user_id))

    await asyncio.gather(*tasks)

    client.close()

if __name__ == '__main__':
    start_time = time.time()

    num_concurrent_users = 10  # 或者 100
    asyncio.run(main(num_concurrent_users))

    end_time = time.time()

    elapsed_time = end_time - start_time
    print(f"Completed {num_concurrent_users} users in {elapsed_time:.2f} seconds.")