使用python脚本来监控进程

简介: 使用python脚本来监控进程

原理:

监控一个指定进程,每隔5秒钟获取其CPU、内存使用量超过60%即kill掉该进程,获取其句柄数,超过300也kill掉该进程

运行环境是windows 64位系统+python 2.7 64位 ,这里需要使用到psutil 类库,要另外安装。脚本里面可以自动安装,前提是你已经下载好了安装包psutil-3.3.0.win-amd64-py2.7.exe

下面看代码:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
from datetime import date, datetime, timedelta
import platform
import os
import win32ui,win32api,win32con,win32gui
import subprocess
def install():
    print("install psutil...")
    sysstr = platform.system()
    if(sysstr =="Windows"):
        print ("Call Windows tasks")
        bit,type=platform.architecture()
        print ("os bit: %s "  % bit)
        #print ("os type: %s "  % type)
        if(bit == "64bit"):
            fileName="psutil-3.3.0.win-amd64-py2.7.exe";
        else:
            fileName="psutil-3.3.0.win32-py2.7.exe";
        print("will install the file [%s]" % fileName)
        
        #启动程序--4种方法
        #subprocess.Popen(fileName); #非阻塞
        #subprocess.Popen(fileName).wati(); #阻塞        
        #os.system(fileName); #阻塞
        #win32api.ShellExecute(0, 'open', fileName, '','',0)
        
        label = 'Setup' #此处假设主窗口名为tt
        hld = win32gui.FindWindow(None, label)        
        count=0
        while (hld == 0 and count<20):
            print("the setup is no running,will run it...")
            count += 1
            win32api.ShellExecute(0, 'open', fileName, '','',0)            
            print("sleep 1 seconds...")
            time.sleep(0.5)
            #wnd = win32ui.GetForegroundWindow()
            #print wnd.GetWindowText()
            hld = win32gui.FindWindow(None, label)
            print("hld is %s" % hld)
        
        pwin=win32ui.FindWindow(None,label)        
        print("pwin is %s" % pwin)
        print pwin.GetWindowText()
        print("click...")
        button2=win32ui.FindWindowEx(pwin,None,None,'下一步(&N) >') #找到按钮 
        button2.SendMessage(win32con.BM_CLICK, 0,-1)
        button2=win32ui.FindWindowEx(pwin,None,None,'下一步(&N) >') #找到按钮 
        button2.SendMessage(win32con.BM_CLICK, 0,-1)
        button2=win32ui.FindWindowEx(pwin,None,None,'下一步(&N) >') #找到按钮 
        button2.SendMessage(win32con.BM_CLICK, 0,-1)
        button2=win32ui.FindWindowEx(pwin,None,None,'完成') #找到按钮 
        button2.SendMessage(win32con.BM_CLICK, 0,-1)
        print("install done...")
        # 鼠标点击
        #print("click...")
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN,0,0) 
        #time.sleep(0.1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP,0,0)
        #time.sleep(1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN,0,0) 
        #time.sleep(0.1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP,0,0)
        #time.sleep(1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN,0,0) 
        #time.sleep(0.1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP,0,0)
        #time.sleep(1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN,0,0) 
        #time.sleep(0.1)
        #win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP,0,0)
        
    elif(sysstr == "Linux"):
        print ("Call Linux tasks")
    else:
        print ("Other System tasks")
        
try:
    print("import psutil...")
    import psutil   
except Exception,e:
    print Exception,":",e
    install()
    import psutil
  
def get_proc_by_id(pid):
    return psutil.Process(pid)
def get_proc_by_name(pname):
    """ get process by name
    
    return the first process if there are more than one
    """
    for proc in psutil.process_iter():
        try:
        # return if found one
        if proc.name().lower() == pname.lower():
            return proc
        except psutil.AccessDenied:
            pass
        except psutil.NoSuchProcess:
            pass
    return None
def getProcess(pname, day=0, hour=0, min=0, second=0):   
    # Init time
    now = datetime.now()
    strnow = now.strftime('%Y-%m-%d %H:%M:%S')
    print "now:",strnow
    # First next run time
    period = timedelta(days=day, hours=hour, minutes=min, seconds=second)
    next_time = now + period
    strnext_time = next_time.strftime('%Y-%m-%d %H:%M:%S')
    print "next run time:",strnext_time
    while True:
        # Get system current time
        iter_now = datetime.now()
        iter_now_time = iter_now.strftime('%Y-%m-%d %H:%M:%S')      
        if str(iter_now_time) == str(strnext_time):
            next_time = iter_now + period
            strnext_time = next_time.strftime('%Y-%m-%d %H:%M:%S')
            print "next run time:",strnext_time
            
            try:
                Process=get_proc_by_name(pname)
            except Exception,e:
              print Exception,":",e
            if Process != None :
                print "-------Found the process : %s" % Process.name();
                print("pid is (%s)" % Process.pid);
                Cpu_usage = Process.cpu_percent(interval=1)
                print("cpu percent is (%s)" % Cpu_usage);
                if (100-Cpu_usage) < 0.1 :
                    print "cpu percent larger 60,now will terminate this process !";
                    Process.terminate();
                    Process.wait(timeout=3);
                    continue
                RAM_percent = Process.memory_percent()
                print("memory percent is (%s)" % RAM_percent);
                if (60-RAM_percent) < 0.1 :
                    print "memory percent larger 60,now will terminate this process !";
                    Process.terminate();
                    Process.wait(timeout=3);
                    continue        
                all_files = list(Process.open_files());
                print("open files size is (%d)" % len(all_files));
                if (len(all_files)>300) :
                    print "open files size larger 300,now will terminate this process !";
                    Process.terminate();
                    Process.wait(timeout=3);
                    continue
                Threads_Num=Process.num_threads()
                print("threads number is (%s)" % Threads_Num); 
                if (Threads_Num>200) :
                    print "threads number larger 200,now will terminate this process !";
                    Process.terminate();
                    Process.wait(timeout=3);
                    continue
            else :
                print "-------No found the process : %s" % pname;
            
            continue
if __name__ == '__main__':
    print("main....")
    getProcess("QQ.exe",second=5)
相关文章
|
6天前
|
安全 Java 数据处理
Python网络编程基础(Socket编程)多线程/多进程服务器编程
【4月更文挑战第11天】在网络编程中,随着客户端数量的增加,服务器的处理能力成为了一个重要的考量因素。为了处理多个客户端的并发请求,我们通常需要采用多线程或多进程的方式。在本章中,我们将探讨多线程/多进程服务器编程的概念,并通过一个多线程服务器的示例来演示其实现。
|
19天前
|
Linux Shell Python
Linux执行Python脚本
Linux执行Python脚本
26 1
|
1月前
|
并行计算 安全 Unix
Python教程第8章 | 线程与进程
本章主要讲解了线程与进程的概念,多线程的运用以及Python进程的相关案例学习
36 0
|
1月前
|
分布式计算 并行计算 Java
浅析Python自带的线程池和进程池
浅析Python自带的线程池和进程池
73 0
|
24天前
|
安全 Python
Python中的并发编程:多线程与多进程技术探究
本文将深入探讨Python中的并发编程技术,重点介绍多线程和多进程两种并发处理方式的原理、应用场景及优缺点,并结合实例分析如何在Python中实现并发编程,以提高程序的性能和效率。
|
8天前
|
JSON 测试技术 持续交付
自动化测试与脚本编写:Python实践指南
【4月更文挑战第9天】本文探讨了Python在自动化测试中的应用,强调其作为热门选择的原因。Python拥有丰富的测试框架(如unittest、pytest、nose)以支持自动化测试,简化测试用例的编写与维护。示例展示了使用unittest进行单元测试的基本步骤。此外,Python还适用于集成测试、系统测试等,提供模拟外部系统行为的工具。在脚本编写实践中,Python的灵活语法和强大库(如os、shutil、sqlite3、json)助力执行复杂测试任务。同时,Python支持并发、分布式执行及与Jenkins、Travis CI等持续集成工具的集成,提升测试效率和质量。
|
15天前
|
存储 监控 异构计算
【Python】GPU内存监控脚本
【Python】GPU内存监控脚本
|
15天前
|
Ubuntu Unix Linux
【Linux/Ubuntu】Linux/Ubuntu运行python脚本
【Linux/Ubuntu】Linux/Ubuntu运行python脚本
|
23天前
|
XML Shell Linux
性能工具之 JMeter 使用 Python 脚本快速执行
性能工具之 JMeter 使用 Python 脚本快速执行
38 1
性能工具之 JMeter 使用 Python 脚本快速执行
|
23天前
|
并行计算 Python
Python中的并发编程:多线程与多进程的比较
在Python编程中,实现并发操作是提升程序性能的重要手段之一。本文将探讨Python中的多线程与多进程两种并发编程方式的优劣及适用场景,帮助读者更好地选择合适的方法来提高程序运行效率。