[Python网络编程]浅析守护进程后台任务的设计与实现
[Python网络编程]浅析守护进程后台任务的设计与实现
在做基于B/S应用中，经常有需要后台运行任务的需求，最简单比如发送邮件。在一些如防火墙，WAF等项目中，前台只是为了展示内容与各种参数配置，后台守护进程才是重头戏。所以在防火墙配置页面中可能会经常看到调用cgi，但真正做事的一般并不是cgi，比如说执行关机命令，他们的逻辑如下：
（ps:上图所说的前台界面包含通常web开发中的后端，不然也没有socket一说)
<喎?http://www.Bkjia.com/kf/ware/vc/" target="_blank" class="keylink">vcD48aDE+ICAgIM6qyrLDtNKq1eLDtMnovMY8L2gxPsTjv8nE3NLJu/POqsqyw7TSqtXiw7TJ6LzGo6zO0r71tcPA7dPJyOfPwqO6ytfPyNPQ0ru148u1w/ejrM/xt8C78Me9tci7+bG+yc+2vNTL0NDU2sDgTGludXjGvcyoyc8gICAgMS6wssirzsrM4iAgY2dp0ruw49Kyvs3TtdPQd3d3yKjP3qOstavWtNDQudi8/LXIw/zB7tDo0qpyb290o6zL+dLU0OjSqsjDuvPMqMrYu6S9+LPMyKW4ySAgICAyLtK7sOPA4MvGt8C78Me9tcS688yoyti7pL34s8zKx0MvQysr0LS1xKOs1NrP+8+iuPHKvcnPuty3vbHjtKbA7aOsyOfM7rPk0ru49s/7z6K94bm5zOW3osvNs/bIpaOsuvPMqL34s8zWu9Do0qrHv9bG16q7u86qtqjS5bXEveG5uczlo6y+zcfhy8m78bXDtKu13bXEss7K/da1oaM8YnIgLz7Ex7/Jsru/ydLUyKW19NbQvOS1xGNpZ8Sjv+mjrNaxvdO3osvNz/vPorj4uvPMqMrYu6S9+LPMxNijv87SvvW1w8rHv8nS1LXEo6yxvs7EtcTW2LXj0rLKx8q1z9bV4rj2t72wuKGjPGJyIC8+PGgxPsjnus7Ktc/WPGJyIC8+PC9oMT7TydPa1+69/NK71rHU2ndpbmRvd3PPwqOsy/nS1M7Sw8e1xMrYu6S9+LPMysfUy9DQ1Np3aW5kb3dzz8K1xKOstavG5Mq1d2luZG93c7Kiw7vT0MrYu6S9+LPMtcS4xcTuo6zP4LbU06a1xMrHt/7O8bXEuMXE7qGj1eLA79Do0qqwstewcHl3aW4zMrD8oaM8cHJlIGNsYXNzPQ=="brush:java;">class MgrService(win32serviceutil.ServiceFramework):
"""
Usage: 'python topmgr.py install|remove|start|stop|restart'
"""
#服务名
_svc_name_ = "Mgr"
#服务显示名称
_svc_display_name_ = "Daemon Mgr"
#服务描述
_svc_description_ = "Daemon Mgr"
def __init__(self, args):
win32serviceutil.ServiceFramework.__init__(self, args)
self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
def SvcDoRun(self):
self.ReportServiceStatus(win32service.SERVICE_START_PENDING)
INFO("mgr startting...")
self.ReportServiceStatus(win32service.SERVICE_RUNNING)
self.start()
# 等待服务被停止
INFO("mgr waitting...")
win32event.WaitForSingleObject(self.hWaitStop, win32event.INFINITE)
INFO("mgr end")
def SvcStop(self):
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
INFO("mgr stopping...")
self.stop()
INFO("mgr stopped")
# 设置事件
win32event.SetEvent(self.hWaitStop)
self.ReportServiceStatus(win32service.SERVICE_STOPPED)
def start(self): pass
def stop(self): pass
很简单，这样就实现了windows中的服务，也就是说脱离终端，运行于后台。INFO等函数只是简单的记录作用，可直接忽略。我们要实现自己的后台程序，只需要继承MgrService，并提供start,stop方法就可以了。
由于我们是通过socket来传递消息的，所以在start方法中要监听端口，等待连接，处理连接，这个大家都很擅长。在这里我选择了 单线程，基于协程，底层使用libev(libevent)--- gevent这个高性能网络库。对gevent有兴趣的童鞋可以看看深度分析gevent运行流程。
class Engine(MgrService):
rbufsize = -1
wbufsize = 0
def start(self):
INFO('wait connection')
self.server = StreamServer((HOST, PORT), self.msg_handle)
self.server.serve_forever()
def msg_handle(self,socket,address):
try:
rfile = socket.makefile('rb', self.rbufsize)
wfile = socket.makefile('wb', self.wbufsize)
headers = Message(rfile).dict
INFO('get a connection from:%s,headers:%s' % (str(address), headers))
if 'module' in headers and headers['module'] in MODULES:
MODULES[headers['module']].handle(wfile, headers)
except Exception:
ERROR('msg_handle exception,please check')
def stop(self):
if hasattr(self, server):
self.server.stop()
当有新连接到来，由msg_handle处理，首先读取发送来的消息，消息格式使用了最简单的http的格式，即（键名:键值）的格式，你要问我为什么采用这个格式，哈哈，格式简单，python有现成的库解析。考虑到后期模块可能很多，所以我们的处理流程自动根据消息的模块参数，调用对应模块的handle方法。
上面代码的那个MODULES是个全局变量，当你添加一个模块的时候需要注册到MODULES中，我提供了module_register方法。
MODULES = { # module: handle module class
}
def module_register(module_name, handle_class):
if module_name in MODULES:
WARN('duplicate module_name:' + module_name)
else:
MODULES[module_name] = handle_class
到这里一切都很自然，但貌似只假设模块有handle方法，自己写一个模块还是很费事，你需要自己去想怎么调用，最有返回什么格式的数据，这都是一件头疼的事情，所以最好提供一个基类模块。
class Module(object):
SECRE_KEY = "YI-LUO-KEHAN"
MODULE_NAME = "BASE_MODULE"
PREFIX = "do_" # method prefix
def __init__(self, wfile, headers):
self.wfile = wfile
self.headers = headers
def __getattr__(self, name):
try:
return self.headers[name]
except Exception:
ERROR("%s has no attr:%s,please check" %(self.MODULE_NAME, name))
@classmethod
def handle(cls, wfile, headers):
module_obj = cls(wfile, headers)
module_obj.schedule_default()
def verify(self):
if hmac.new(self.SECRE_KEY, self.MODULE_NAME).hexdigest() == self.signature:
return True
else:
WARN("client verify failed,signature:%s" % str(self.signature))
def schedule_default(self):
err_code = 0
if self.verify() and self.action:
func_name = self.PREFIX + self.action
try:
getattr(self, func_name)()
except AttributeError:
err_code = 1
ERROR("%s has no method:%s" %(self.MODULE_NAME, func_name))
except Exception:
err_code = 2
ERROR("module:%s,method:%s,exception" % (self.MODULE_NAME, func_name))
else:
err_code = 3
if err_code:
self.send_error({'err_code':err_code})
def send_success(self, msg=''):
data = {'success':True,'msg':msg}
self.wfile.write(json.dumps(data))
def send_error(self, msg=''):
data = {'success':False,'msg':msg}
self.wfile.write(json.dumps(data))在基类模块中我们提供了默认的处理流程，即根据消息中action,调用do_action方法，并提供了一个简单但很有效的认证方法，通过消息的signature字段，可能有些简陋，但没关系，你可以定义自己的认证方法。
下面该写我们自己的模块了，
TASK = {} # task_id: pid
class ScanModule(Module):
MODULE_NAME = "SCAN_MODULE"
def do_start(self):
self.send_success('start ok')
DEBUG('------------task start------------')
task_ids = [int(task_id) for task_id in self.task_ids.split(',') if int(task_id) not in TASK]
for task_id in task_ids:
try:
cmd = 'python scan.py -t %s' % task_id
DEBUG(cmd)
self.sub = Popen(cmd, shell=True, cwd=CWD)
pid = int(self.sub.pid)
TASK[task_id] = pid
INFO('%s start a new task,task_id:%s,pid:%s' %(self.MODULE_NAME, task_id, pid))
except Exception:
ERROR('%s start a new task,task_id:%s failed' % (self.MODULE_NAME, task_id))
def do_stop(self):
self.send_success('stop ok')
DEBUG('------------task stop------------')
task_ids = [int(task_id) for task_id in self.task_ids.split(',') if int(task_id) in TASK]
for task_id in task_ids:
pid = TASK.pop(task_id)
try:
INFO('%s stop a new task,task_id:%s,pid:%s' %(self.MODULE_NAME, task_id, pid))
call(['taskkill', '/F', '/T', '/PID', str(pid)])
except Exception:
ERROR('%s taskkill a task failed,task_id:%s,pid:%s' %(self.MODULE_NAME, task_id, pid))
module_register(ScanModule.MODULE_NAME, ScanModule)
上面实现了一个简单的扫描模块，支持两个action,start,stop。start很简单，调用gevent的subprocess.Popen运行子进程，并记录pid，stop则使用taskkill直接杀掉该进程。这里有两点需要注意： 1.不要用原生的subprocess模块，因为原生的subprocess是阻塞的，这可能导致主处理逻辑也阻塞，不能服务更多的请求最后别忘了调用module_register注册相应模块。 2.方法一开始最好就返回结果，因为前台很可能在等待返回。所以说as soon as possible下面提供一个客户端用于测试，client.py
#!/usr/bin/env python
#-*-encoding:UTF-8-*-
import hmac
import gevent
from gevent import monkey
monkey.patch_socket()
addr = ('localhost', 6667)
def send_request(module_name,request_headers):
SECRE_KEY = "YI-LUO-KEHAN"
socket = gevent.socket.socket()
socket.connect(addr)
request_headers['module'] = module_name
request_headers['signature'] = hmac.new(SECRE_KEY, module_name).hexdigest()
h = ["%s:%s" %(k, v) for k,v in request_headers.iteritems()]
h.append('\n')
request = '\n'.join(h)
socket.send(request)
print socket.recv(8192)
socket.close()
if __name__ =="__main__":
import sys
if sys.argv[1] == 'start':
send_request('SCAN_MODULE',{'action':'start','task_ids':'1'})
else:
send_request('SCAN_MODULE',{'action':'stop','task_ids':'1'})
我们来简单的测试一下：注意：由于要注册到服务，cmd需要管理员权限至于start中调用的scan.py随便写一个就可以
截图如下，我们看到成功！！！
本文代码已放到github，https://github.com/Skycrab/pymgr感兴趣的童鞋可以参考，请大家多提意见。
评论关闭