一 背景
现实总有些奇葩的需求,比如你需要占用 50%的 cpu，或者你需要占用 80%的 cpu 的程序，这个程序没有其他作用，仅仅是为了占用 cpu，空跑，但是其实还隐藏一个需求就是不能影响正常的程序的运行，所以最好要有动态调节的功能。
应用场合多是，现场申请的资源多了，甲方又要根据资源情况来回收机器，没办法，只能采用这个妥协的办法了。
二 核心算法
逻辑还是很简单，就是判断系统现在的负载，如果超过需要的负载，则不需要运行休眠即可，如果需要运行就进行计算占用 cpu，根据需要的占用比例占用一段时间即可。
用 c 去写有点麻烦，刚好网上找到了别人开源的 python 代码，就试验下分享给大家。
核心算法代码如下，random.random()获取一个 0-1 之间的随机小数，*100以后就成了 0-100 之间的数字，target 标识我们需要占用 cpu 的百分比，比如需要占用总体 cpu 的 50%，那么target的值就是 50，multiplier是总的 cpu 数量；
假如我们在 2 个 cpu 的环境下，需要占用 50%的 cpu，则target * multiplier为 100，这段代码就占 100%的 cpu，即 1 个 cpu，总体 cpu 的占用率为 50%，这段代码则在 0.01s 的时间内完全占用一个 cpu；如果需要占用 30%的 cpu，则target * multiplier为 60，则在 60%的时间内是占用 cpu，在 40%的时间内休眠，则不占用 cpu，上面再套个函数控制下时间就可以满足要求了。
@staticmethod
def my_kernel(target, multiplier):
""" CPU kernel
"""
rand = 100 * random.random()
if rand < target * multiplier:
start = time.time()
while time.time() - start < 0.01:
rand ** 3
else:
time.sleep(0.01)
套个函数来实现 10s 内基本满足 cpu 占用符合我们的预期。
def run_awhile(self, sec=10):
start = time.time()
while time.time() - start < sec:
self.my_kernel(self.target, self.multiplier)
三 其他关键点
3.1 python 占用多 cpu
python 来说如果需要占用多个 cpu，不能用多线程，只能用多进程来占用，可以通过继承multiprocessing.Process类来实现。
3.2 python 更改进程名
python 更改进程名，可以通过安装setproctitle模块来控制，更改进程名如下，这样可以更好的伪装了：）。
import setproctitle
setproctitle.setproctitle("forge_load_cpu")
3.3 python 获取 cpu 个数和负载
cpu 个数和 cpu 的占用情况，可以通过模块psutil，如下：
#获取cpu个数
psutil.cpu_count()
#获取cpu占用率
psutil.cpu_percent()
cpu 的使用率比较特殊，因为是实时变化的，我们最好取平均值，作者实现的比较巧妙，通过双向队列来保存 cpu 占用率，获取的时候，再通过取平均值的办法，让获取的信息更加准确。
实现代码如下：
class Monitor(threading.Thread):
""" 后台检测当前GPU占用率
"""
def __init__(self):
super(Monitor, self).__init__()
self.setDaemon(True)
self._queue = deque([0] * 10, 10)
self.avg_load = 0
self.max_load = 0
def update(self, ):
load = self.get_current_load()
self._queue.append(load)
self.avg_load = sum(self._queue)/len(self._queue)
self.max_load = max(self._queue)
def run(self):
while True:
self.update()
time.sleep(0.3)
@staticmethod
def get_current_load():
return psutil.cpu_percent()
四 试用下
4.1 软件安装
#安装pip 用于安装python模块
yum -y install python-pip
#升级
pip install --upgrade pip
#安装两个依赖库
pip install psutil
#安装不了可以下载安装
#wget https://pypi.python.org/packages/source/p/psutil/psutil-2.1.3.tar.gz
#tar xvf psutil-2.1.3.tar.gz
# cd psutil-2.1.3;make &&make install
pip install setproctitle
4.2 设置 cpu 占用和运行
程序通过获取环境变量 TARGET 来得到需要占用 cpu 的比例，默认是 50，即占用 50%的 cpu。
export TARGET=70
python cpu.py
如下图：
为了让 cpu 占用更精确，那只有把核心算法里面的时间片设置小，上图为 0.01，运行不是很精确，设置为 0.001，运行才更精确些。
整体代码如下：
# -*- coding: utf-8 -*-
#!/usr/bin/python
##################################################
# AUTHOR : Yandi LI
# CREATED_AT : 2018-11-01
# LAST_MODIFIED : 2018-11-12 15:46:55
# USAGE : python -u main.py
# PURPOSE : GPU占用程序
##################################################
from __future__ import division
import random
import threading
import multiprocessing
import time
from collections import deque
import psutil
CPU_COUNT = psutil.cpu_count()
class Monitor(threading.Thread):
""" 后台检测当前GPU占用率
"""
def __init__(self):
super(Monitor, self).__init__()
self.setDaemon(True)
self._queue = deque([0] * 10, 10)
self.avg_load = 0
self.max_load = 0
def update(self, ):
load = self.get_current_load()
self._queue.append(load)
self.avg_load = sum(self._queue)/len(self._queue)
self.max_load = max(self._queue)
def run(self):
while True:
self.update()
time.sleep(0.3)
@staticmethod
def get_current_load():
return psutil.cpu_percent()
class Worker(multiprocessing.Process):
""" CPU占用程序
- 根据目标target，自动调整需要用到的CPU核心数量
- 如果monitor检测有其他程序争抢CPU，峰值超过阈值，则自动切断运行
"""
def __init__(self, target=50):
super(Worker, self).__init__()
self.target = target
self.multiplier = 1
self.daemon = True
@staticmethod
def my_kernel(target, multiplier):
""" CPU kernel
"""
rand = 100 * random.random()
if rand < target * multiplier:
start = time.time()
while time.time() - start < 0.001:
rand ** 3
else:
time.sleep(0.001)
def run_awhile(self, sec=10):
start = time.time()
while time.time() - start < sec:
self.my_kernel(self.target, self.multiplier)
def idle_awhile(self, sec=5):
time.sleep(sec)
def _boost(self, rate=1.05):
self.multiplier *= rate
def _slow_down(self, rate=1.1):
self.multiplier /= rate
def adjust_speed(self, avg_load):
if avg_load < self.target * 0.8:
self._boost()
# print("Adjusted speed: boost")
return
if avg_load > self.target * 1.05:
self._slow_down()
# print("Adjusted speed: slow_down")
return
def run(self):
monitor = Monitor()
monitor.start()
print("Monitor started: %s" % monitor.is_alive())
time.sleep(5)
print("Initial average load", monitor.avg_load)
while True:
if monitor.max_load > self.target * 1.1:
sec = random.random() * 3 + 1
# print("Idle for %ss with max_load %s, avg_load %s" % (sec, monitor.max_load, monitor.avg_load))
self.idle_awhile(sec)
continue
sec = random.random() * 3 + 1
# print("Run for %ss with avg_load %s and multiplier %s" % (sec, monitor.avg_load, self.multiplier))
self.run_awhile(sec)
self.adjust_speed(monitor.avg_load)
if __name__ == "__main__":
import os
import setproctitle
setproctitle.setproctitle("forge_load_cpu")
target = float(os.environ.get("TARGET", 55))
workers = []
for i in range(CPU_COUNT):
worker = Worker(target)
worker.start()
print("Worker %d started: %s" % (i, worker.is_alive()))
workers.append(worker)
for worker in workers:
worker.join()
作者 github 开源地址：
https://github.com/yandili/forge_load/blob/master/cpu_load/README.md
五 诗词欣赏
江城子·密州出猎
--苏轼
老夫聊发少年狂，左牵黄，右擎苍，锦帽貂裘，千骑卷平冈。
为报倾城随太守，亲射虎，看孙郎。
酒酣胸胆尚开张，鬓微霜，又何妨！
持节云中，何日遣冯唐？
会挽雕弓如满月，西北望，射天狼。