MongoDB Test7: 并发测试

需求

模块

代码

import time
import concurrent.futures
from pymongo import MongoClient
from gridfs import GridFS

# MongoDB连接信息
mongo_host = '192.168.92.159'
mongo_port = 20000
database_name = 'test_db'

# 文件路径
file_path = '.\data\data718.segy'
# 并发用户数
concurrent_users = [10, 100]

def upload_file(file_path):
    client = MongoClient(mongo_host, mongo_port)
    db = client[database_name]
    fs = GridFS(db)

    start_time = time.time()
    with open(file_path, 'rb') as file:
        fs.put(file, filename='uploaded_file')
    end_time = time.time()

    client.close()
    return end_time - start_time

def download_file(file_path):
    client = MongoClient(mongo_host, mongo_port)
    db = client[database_name]
    fs = GridFS(db)

    start_time = time.time()
    with open('downloaded_file.txt', 'wb') as file:
        grid_out = fs.get_last_version(filename='uploaded_file')
        file.write(grid_out.read())
    end_time = time.time()

    client.close()
    return end_time - start_time

def test_concurrency(num_users):
    print(f'并发用户数：{num_users}')
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_users) as executor:
        upload_times = list(executor.map(upload_file, [file_path] * num_users))
        download_times = list(executor.map(download_file, [file_path] * num_users))

    avg_upload_time = sum(upload_times) / num_users
    avg_download_time = sum(download_times) / num_users

    print(f'平均上传时间：{avg_upload_time:.4f} 秒')
    print(f'平均下载时间：{avg_download_time:.4f} 秒')

if __name__ == '__main__':
    for num_users in concurrent_users:
        test_concurrency(num_users)

测试结果