CodePattern/B 高性能模式/util.py

import time
import psutil
import tracemalloc
import requests
import aiohttp


def compute_task():
    start_time = time.time()
    result = sum(i * i for i in range(10**7))
    end_time = time.time()
    print(f"compute_task耗时：{end_time - start_time}秒")


async def async_compute_task():
    start_time = time.time()
    result = sum(i * i for i in range(10**7))
    end_time = time.time()
    print(f"async_compute_task耗时：{end_time - start_time}秒")


def measure_performance(func):

    def wrapper(*args, **kwargs):
        print(f"开始运行{func.__name__}")

        time.sleep(30)
        print(f"运行前的CPU使用率：{psutil.cpu_percent()}%")
        print(f"运行前的内存使用率：{psutil.virtual_memory().percent}%")
        print()

        tracemalloc.start()
        start_time = time.time()
        func(*args, **kwargs)
        end_time = time.time()
        current, peak = tracemalloc.get_traced_memory()
        tracemalloc.stop()
        print(f"{func.__name__}耗时：{end_time - start_time}秒")
        print(f"当前内存使用：{current / 10**6}MB, 峰值内存使用：{peak / 10**6}MB")
        print(f"CPU使用率：{psutil.cpu_percent(interval=1)}%")
        print(f"内存使用率：{psutil.virtual_memory().percent}%")
        print()

        time.sleep(30)
        print(f"运行结束30秒后的CPU使用率：{psutil.cpu_percent()}%")
        print(f"运行结束30秒后的内存使用率：{psutil.virtual_memory().percent}%")
        print(f"运行结束{func.__name__}结束")
        print()

    return wrapper


def measure_performance_async(func):

    async def wrapper(*args, **kwargs):
        print(f"开始运行{func.__name__}")

        time.sleep(30)
        print(f"运行前的CPU使用率：{psutil.cpu_percent()}%")
        print(f"运行前的内存使用率：{psutil.virtual_memory().percent}%")
        print()

        tracemalloc.start()
        start_time = time.time()
        await func(*args, **kwargs)
        end_time = time.time()
        current, peak = tracemalloc.get_traced_memory()
        tracemalloc.stop()
        print(f"{func.__name__}耗时：{end_time - start_time}秒")
        print(f"当前内存使用：{current / 10**6}MB, 峰值内存使用：{peak / 10**6}MB")
        print(f"CPU使用率：{psutil.cpu_percent(interval=1)}%")
        print(f"内存使用率：{psutil.virtual_memory().percent}%")
        print()

        time.sleep(30)
        print(f"运行结束30秒后的CPU使用率：{psutil.cpu_percent()}%")
        print(f"运行结束30秒后的内存使用率：{psutil.virtual_memory().percent}%")

    return wrapper


def fetch_url(url):
    response = requests.get(url)


async def fetch_url_async(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, ssl=False) as response:
            await response.read()


urls = [
    "https://www.baidu.com", "https://www.qq.com", "https://www.taobao.com",
    "https://www.tmall.com", "https://www.jd.com", "https://www.sina.com.cn",
    "https://www.weibo.com", "https://www.163.com", "https://www.sohu.com",
    "https://www.ifeng.com", "https://www.alipay.com", "https://www.youku.com",
    "https://www.iqiyi.com", "https://www.tudou.com", "https://www.douyin.com",
    "https://www.meituan.com", "https://www.xiaomi.com",
    "https://www.oppo.com", "https://www.vivo.com", "https://www.dianping.com",
    "https://www.autohome.com.cn", "https://www.hupu.com",
    "https://www.xinhuanet.com", "https://www.chinadaily.com.cn",
    "https://www.people.com.cn", "https://www.cctv.com", "https://www.cntv.cn",
    "https://www.zol.com.cn", "https://www.pconline.com.cn",
    "https://www.techweb.com.cn", "https://www.cs.com.cn",
    "https://www.eastmoney.com", "https://www.hexun.com",
    "https://www.yicai.com", "https://www.21jingji.com"
]