Locust（俗称 蝗虫）一个轻量级的开源压测工具，基本功能是用Python代码描述所有测试。不需要笨拙的UI或庞大的XML，只需简单的代码即可。
有一段时间没有弄性能测试了，最近需要压一压性能。 有人用python去调用接口，然后用chales抓包，导出成har. 然后将har转化成jmx，然后导入jemeter做压测。我觉得这样太麻烦。 既然都是用python,为何不用python支持的locust. 于是我将locust捡起来。网上的资料太少，太零散，而且好多都过时了，就是错的。现在写这个，就是做一个记录，少一些坑。
安装和使用
代码语言：javascript复制pip install locust
你看到的很多文档，都是老的，包早就改了。 locust版本0.13之后已经废除了min_wait和max_wait的使用
代码语言：javascript复制 min_wait = 3000
max_wait = 7000
改为使用
代码语言：javascript复制wait_time = between(3,7)
而且这个类已经废弃了： HttpLocust 换成：HttpUser 一直报错，说没有task, 后面发现， task_set = UserBehavior 改成 tasks = [UserBehavior]
增加断言
对于测试来说，每个自动化测试用例都应该有个断言判断，这样才能知道测试用例的成功/失败。
在Python Locust中，通过with的方式来增加断言。具体脚本如下：
代码语言：javascript复制with self.client.post(url=url, data=params, timeout=10, catch_response=True) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
注意：catch_response=True这个参数是必须要加的，否则在性能测试时，后台会一直报错，提示AttributeError: 'Response' object has no attribute 'success'。
刚刚的样例中，是举例断言status_code是否等于200，如果是，则返回成功，反之返回失败。
思考一下，我们运行性能测试的时候，需要哪些数据呢？ step1 准备脚本过程数据 1.全局数据：exp(一批测试用户、一批测试商品) 2.局部数据： a) 用于接口动态入参，不用从上下文中获取 (exp：时间戳) b) 用于接口动态入参，需要从上下文中获取（exp：用户登录 token 或者 cookie,订单 ID） 3.转换数据：exp(将明文密码转换为 md5 加密) setp2 运行脚本过程中数据使用 1.同一批数据：每个接口可重复取数 2.同一批数据，每个接口取数不重复 3.不同批数据：每个接口分别随机取值，不同批数据无关联关系，可随机匹配 4.不同批数据：每个接口分别随机取值，不同批数据有对应关联关系 二、关于上述场景数据参数化 step1 准备脚本过程中参数化处理 1.全局数据：（对比 jmeter 调用 text 或者 csv 的文档方法） 可以使用 txt 或者 csv,在通过 pyton 中文件读取函数读取出来，有数据关联的建议每条以 dict of list 形式进行存储 使用 python 中 list 类型读取 2.局部数据： a.上下文无关的动态参数，使用 def 函数返回值进行动态获取 b.上下文相关的动态参数，使用 taskset 中类属性来进行获取并参数化
代码语言：javascript复制# txt中数据为1,2,3,4,5,6
def get_txt_data():
with open("user.txt", "r") as f:
data = f.read()
return data
# 列表形式数据
s = [1, 2, 3, 4, 5, 6]
s1 = [{"user_name": "张三", "password": 123456},
{"user_name": "李四", "password": 456789},
{"user_name": "王五", "password": 123789}]
# 从上下文中取数据
from locust import Locust, TaskSet, task, between
from time import time
from uuid import uuid1
# 返回当前时间戳
def get_time_stamp():
return str(int(time()))
class MyTaskSet(TaskSet):
# 该函数不是task任务的，可以添加函数来进行获取token,方便任务来调用
def get_login_token(self):
return uuid1()
# 使用类属性来进行参数传递
token = None
# TaskSet相当于下面所有task的大脑
@task(1) # 声明任务
def my_task(self):
print("执行task1" get_txt_data())
MyTaskSet.token = uuid1()
@task(2)
def my_task_2(self):
print("执行task2" str(s[0]) str(MyTaskSet.token))
class AppUser(Locust):
weight = 1 # 赋值权重 weight = 10 默认为10
task_set = MyTaskSet
wait_time = between(1, 2)
host = "" # 域名host
step2 脚本运行过程中 1.同一批数据或者多批数据： 不同的 taskset 对应的 user 类中，使用相同的 data 数据 不同的 taskset 对应的 user 类中，通过一个或者多个队列进行不重复取值 (如果有关联关系的数据，可以重新一个队列，把对应关系用字典存入到队列)
代码语言：javascript复制# 两个UserLocust重复调用一个list数据
from locust import HttpLocust, TaskSet, task, Locust, between
import queue
# from multiprocessing import Queue
s=[1,2,3,4,5,6]
class MyTaskSet(TaskSet):
@task()
def task_test1(self):
print(queue_data_test1[0])
class MyTaskSet2(TaskSet):
@task()
def task_test2(self):
print(queue_data_test2[0])
class UserLocust2(Locust):
task_set = MyTaskSet2
wait_time = between(3, 10)
queue_data_test2 =s
class UserLocust(Locust):
task_set = MyTaskSet
wait_time = between(3, 10)
queue_data_test1 = s
# 两个UserLocust调用同一个队列，数据取值不重复;（存在多批次不重复，可定义多个Queue()对象）
from locust import HttpLocust, TaskSet, task, Locust, between
import queue
# from multiprocessing import Queue
def queue_data():
queue_data = queue.Queue() # 默认为先进先出 该队列为task_set共享
# queue.LifoQueue(),后进先出
for i in range(0, 13):
queue_data.put_nowait(i) # put_nowait 不阻塞
return queue_data
class MyTaskSet(TaskSet):
@task()
def task_test1(self):
print(self.locust.queue_data_test1.get())
class MyTaskSet2(TaskSet):
@task()
def task_test2(self):
print(self.locust.queue_data_test2.get())
class UserLocust2(Locust):
weight =1
task_set = MyTaskSet2
wait_time = between(3, 10)
queue_data_test2 = queue_data() # 默认为先进先出 该队列为task_set共享
class UserLocust(Locust):
weight =3
task_set = MyTaskSet
wait_time = between(3, 10)
queue_data_test1 = queue_data()
三.总结： 上面只是用了样例描述了一下 locust 参数化的过程，基本上是用 python 的方法或者类调用来获取数据和传递参数 选择使用 Queue() 对象来实现数据的不重复调用，从其他资料中也可以看到使用 from multiprocessing import Queue 的队列对象来控制多进程安全数据传递
如果我们写性能测试的case,如果一个一个写，得写半天。 我自己写了个方法，自动生成脚本。 首先得用httprunner来生成一个yaml文件，像这样：
代码语言：javascript复制teststeps:
- name: /services/api/mobile/service/appversion
request:
headers:
Accept-Encoding: gzip
Connection: Keep-Alive
Content-Length: '77'
Content-Type: application/json; charset=UTF-8
Host: stg-ec.ef.com.cn
User-Agent: Android/SmartEnglish:2.1.14(okhttp/4.8.1)
json:
serviceRequest:
appVersion: 2.1.14
platform: Android
productId: 4
method: POST
url: https://stg-ec.ef.com.cn/services/api/mobile/service/appversion
validate:
- eq:
- status_code
- 200
- eq:
- headers.Content-Type
- application/json; charset=utf-8
然后将这个yaml文件生成locust可以压的性能测试脚本。 脚本如下：
代码语言：javascript复制from apitest import get_test_data
import pytest
import requests
import allure
cases, parameters = get_test_data('C:\work\code\python\oneapp.yaml')
list_params = list(parameters)
def test_each_api():
with open("C:\work\code\python\locus.py","w ") as f:
f.write("from locust import HttpUser, TaskSet, task n")
f.write("from locust import between n")
f.write("class UserTask(TaskSet):n")
for i in list_params:
f.write("t@taskn")
f.write("tdef {}(self):n".format(i[0].split("/")[-1]))
f.write("ttheader = {}n".format(i[1]['headers']))
f.write("ttjson = {}n".format(i[1]['json']))
f.write("ttjson['serviceRequest']['token'] = tokenn")
f.write("ttjson['serviceRequest']['sessionId'] = sessionn")
f.write(
"ttwith self.client.pos(url='{}', headers=header, json=json, verify=False,timeout=10, catch_response=True) as response:n".format(i[0]))
f.write("tttif response.status_code == 200:n")
f.write("ttttresponse.success()n")
f.write("tttelse:n")
f.write("ttttresponse.failure('Failed!')n")
f.write("class User(HttpUser):n")
f.write("ttasks = [UserTask]n")
f.write("twait_time = between(3,7)n")
test_each_api()
生成的文件如下：
代码语言：javascript复制from locust import HttpUser, TaskSet, task
from locust import between
class UserTask(TaskSet):
token=None
session=None
def on_start(self):
self.login()
# 随机返回登录用户
def login_user(self):
users = {"username": 's3035'}
username = users["username"]
password = 1
return username, password
@task
def login(self):
username, password = self.login_user()
url = "/services/api/mobile/service/login"
json = {
"serviceRequest": {
"userName": username,
"password": password,
"appVersion": "2.1.14",
"platform": "Android",
"productId": 4
}
}
res = self.client.post(url=url, json=json, verify=False)
token = res.json()["serviceResponse"]["token"]
session = res.json()["serviceResponse"]["sessionId"]
UserTask.token=token
UserTask.session=session
@task
def appversion(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '77',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'appVersion': '2.1.14', 'platform': 'Android', 'productId': 4}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/appversion', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def ec_resetpassword(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '294',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {
'serviceRequest': {'appVersion': '2.1.14', 'culturecode': 'en', 'platform': 'Android', 'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'type': 'force', 'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/ec_resetpassword', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def ec_studycontext(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '279',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {
'serviceRequest': {'appVersion': '2.1.14', 'culturecode': 'en', 'platform': 'Android', 'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/ec_studycontext', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def ec_enrolledcourses(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '345',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'appVersion': '2.1.14', 'countrycode': 'cn', 'courseTypes': ['GE', 'BE'],
'culturecode': 'en', 'partnercode': 'Socn', 'platform': 'Android',
'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/ec_enrolledcourses', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def ec_studyplanhome_phoenix(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '340',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'appVersion': '2.1.14', 'countrycode': 'cn', 'culturecode': 'en',
'partnercode': 'Socn',
'platform': 'Android', 'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'siteVersion': '18-1',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/ec_studyplanhome_phoenix', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def switchgetuiclienttoken(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '398',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {
'serviceRequest': {'appVersion': '2.1.14', 'bind': True, 'clientId': 'b6659c428be9bc8f8066074ee612624f',
'countrycode': 'cn', 'culturecode': 'en', 'partnercode': 'Socn',
'platform': 'Android',
'productId': 4, 'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'siteVersion': '18-1',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/switchgetuiclienttoken', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def ec_getlegaldisclaimer(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '340',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'appVersion': '2.1.14', 'countrycode': 'cn', 'culturecode': 'en',
'partnercode': 'Socn',
'platform': 'Android', 'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'siteVersion': '18-1',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/ec_getlegaldisclaimer', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def legalpage(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '340',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'appVersion': '2.1.14', 'countrycode': 'cn', 'culturecode': 'en',
'partnercode': 'Socn',
'platform': 'Android', 'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'siteVersion': '18-1',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/legalpage', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def banner(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '340',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'appVersion': '2.1.14', 'countrycode': 'cn', 'culturecode': 'en',
'partnercode': 'Socn',
'platform': 'Android', 'productId': 4,
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'siteVersion': '18-1',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA',
'unifiedLogin': True}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/mobile/service/banner', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def resourcemap(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '332',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'body': {'dataStore': 'cn1', 'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA'},
'header': {'appVersion': '2.1.14', 'cultureCode': 'en', 'lastUpdate': 0,
'platform': 'Android', 'productId': 3, 'unifiedLogin': True}}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/ecmobile/service/resourcemap', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def studentcontext(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '332',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'body': {'dataStore': 'cn1', 'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA'},
'header': {'appVersion': '2.1.14', 'cultureCode': 'en', 'lastUpdate': 0,
'platform': 'Android', 'productId': 3, 'unifiedLogin': True}}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/ecmobile/service/studentcontext', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
@task
def systemmessage(self):
header = {'Accept-Encoding': 'gzip', 'Connection': 'Keep-Alive', 'Content-Length': '393',
'Content-Type': 'application/json; charset=UTF-8', 'Host': 'stg-ec.ef.com.cn',
'User-Agent': 'Android/SmartEnglish:2.1.14(okhttp/4.8.1)'}
json = {'serviceRequest': {'body': {'countryCode': 'cn', 'dataStore': 'cn1', 'partnerCode': 'Socn',
'sessionId': '374d09bfcd3045d5bc71be36ecbe2e87', 'siteVersion': '18-1',
'token': 'ADIAMAAwADAAMAAxADIANwA2AHwAMQA2ADEAOQA1ADkAMgAzADQAMQA2ADUAMQB8AHMAdABlAHMAdABjADEAMwAwADcAOAB8AE0AfABTA'},
'header': {'appVersion': '2.1.14', 'cultureCode': 'en', 'lastUpdate': 0,
'platform': 'Android', 'productId': 3, 'unifiedLogin': True}}}
json['serviceRequest']['token'] = UserTask.token
json['serviceRequest']['sessionId'] = UserTask.session
with self.client.post(url='/services/api/ecmobile/service/systemmessage', headers=header, json=json,
verify=False) as response:
if response.status_code == 200:
response.success()
else:
response.failure('Failed!')
class User(HttpUser):
tasks = [UserTask]
wait_time = between(3,7)
host = "https://com.cn"
现在就可以进行压测了。 Teminal里面输入： locust -f locust4.py --host=https://xxx.com" 然后： 在浏览器中输入：http://localhost:8089/ 效果如下：
曲线如下：
no-web模式启动：os.system(“locust -f locust4.py --host=https://www.baidu.com --no-web --csv=example -c 100 -r 10 -t 10s”)
这样就可以压测了。 还可以这样监控： psutil是一个开源跨平台的库，其提供了便利的函数用来获取系统的信息，比如CPU，内存，磁盘，网络等。此外，psutil还可以用来进行进程管理，包括判断进程是否存在、获取进程列表、获取进程详细信息等。而且psutil还提供了许多命令行工具提供的功能，包括：ps，top，lsof，netstat，ifconfig， who，df，kill，free，nice，ionice，iostat，iotop，uptime，pidof，tty，taskset，pmap。
安装：
代码语言：javascript复制pip install psutil
可以这样用：
代码语言：javascript复制CPUprint(psutil.cpu_count()) #查看CPU逻辑数量
print(psutil.cpu_count(logical=False)) #查看CPU物理核心
print(psutil.cpu_percent()) #查看CUP利用率
内存
print(psutil.virtual_memory()) #查看物理内存
print(psutil.swap_memory()) #查看交换内存
磁盘
print(psutil.disk_usage('/')) # 磁盘使用情况
print(psutil.disk_partitions()) #磁盘分区信息
print(psutil.disk_io_counters()) #磁盘IO
进程
print(psutil.pids()) #获取所有进程ID
print(psutil.Process(492).name()) #获取ID为492的进程的进程名
模拟ps的效果
print(psutil.test()) #模拟ps命令的效果
一个实时打印系统CPU使用率，内存使用率的例子
代码语言：javascript复制import psutil
import time
print("CPU使用率 内存使用率")
delay = 1
while True:
time.sleep(delay)
print(str(psutil.cpu_percent()) "% " str(psutil.virtual_memory().percent) "% ")
这样用起来就很爽了。
