Python使用Redis实现一个简单作业调度系统

概述

Redis作为内存数据库的一个典型代表，已经在很多应用场景中被使用，这里仅就Redis的pub/sub功能来说说怎样通过此功能来实现一个简单的作业调度系统。这里只是想展现一个简单的想法，所以还是有很多需要考虑的东西没有包括在这个例子中，比如错误处理，持久化等。

下面是实现上的想法

MyMaster：集群的master节点程序，负责产生作业，派发作业和获取执行结果。

MySlave：集群的计算节点程序，每个计算节点一个，负责获取作业并运行，并将结果发送会master节点。

channel CHANNEL_DISPATCH：每个slave节点订阅一个channel，比如“CHANNEL_DISPATCH_[idx或机器名]”，master会向此channel中publish被dispatch的作业。

channel CHANNEL_RESULT：用来保存作业结果的channel，master和slave共享此channel，master订阅此channel来获取作业运行结果，每个slave负责将作业执行结果发布到此channel中。

Master代码

	#!/usr/bin/env python
02	# -*- coding: utf-8 -*-
03	import time
04	import threading
05	import random
06	import redis
07	 
08	 
09	REDIS_HOST = 'localhost'
10	REDIS_PORT = 6379
11	REDIS_DB = 0
12	CHANNEL_DISPATCH = 'CHANNEL_DISPATCH'
13	CHANNEL_RESULT = 'CHANNEL_RESULT'
14	 
15	 
16	class MyMaster():
17	    def __init__(self):
18	        pass
19	 
20	    def start(self):
21	        MyServerResultHandleThread().start()
22	        MyServerDispatchThread().start()
23	 
24	 
25	class MyServerDispatchThread(threading.Thread):
26	    def __init__(self):
27	        threading.Thread.__init__(self)
28	 
29	    def run(self):
30	        r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
31	        for i in range(1, 100):
32	            channel = CHANNEL_DISPATCH + '_' +str(random.randint(1, 3))
33	            print("Dispatch job %s to %s" % (str(i), channel))
34	            ret = r.publish(channel, str(i))
35	            if ret == 0:
36	                print("Dispatch job %s failed." % str(i))
37	            time.sleep(5)
38	 
39	 
40	class MyServerResultHandleThread(threading.Thread):
41	    def __init__(self):
42	        threading.Thread.__init__(self)
43	 
44	    def run(self):
45	        r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
46	        p = r.pubsub()
47	        p.subscribe(CHANNEL_RESULT)
48	        for message in p.listen():
49	            if message['type'] != 'message':
50	                continue
51	            print("Received finished job %s" % message['data'])
52	 
53	 
54	if __name__ == "__main__":
55	    MyMaster().start()
56	    time.sleep(10000)

说明

MyMaster类 – master主程序，用来启动dispatch和resulthandler的线程

MyServerDispatchThread类 – 派发作业线程，产生作业并派发到计算节点

MyServerResultHandleThread类 – 作业运行结果处理线程，从channel里获取作业结果并显示

Slave代码

	#!/usr/bin/env python
02	# -*- coding: utf-8 -*-
03	from datetime import datetime
04	import time
05	import threading
06	import random
07	import redis
08	 
09	 
10	REDIS_HOST = 'localhost'
11	REDIS_PORT = 6379
12	REDIS_DB = 0
13	CHANNEL_DISPATCH = 'CHANNEL_DISPATCH'
14	CHANNEL_RESULT = 'CHANNEL_RESULT'
15	 
16	 
17	class MySlave():
18	    def __init__(self):
19	        pass
20	 
21	    def start(self):
22	        for i in range(1, 4):
23	            MyJobWorkerThread(CHANNEL_DISPATCH + '_' +str(i)).start()
24	 
25	 
26	class MyJobWorkerThread(threading.Thread):
27	 
28	    def __init__(self, channel):
29	        threading.Thread.__init__(self)
30	        self.channel = channel
31	 
32	    def run(self):
33	        r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
34	        p = r.pubsub()
35	        p.subscribe(self.channel)
36	        for message in p.listen():
37	            if message['type'] != 'message':
38	                continue
39	            print("%s: Received dispatched job %s " %(self.channel, message['data']))
40	            print("%s: Run dispatched job %s " % (self.channel, message['data']))
41	            time.sleep(2)
42	            print("%s: Send finished job %s " % (self.channel, message['data']))
43	            ret = r.publish(CHANNEL_RESULT, message['data'])
44	            if ret == 0:
45	                print("%s: Send finished job %s failed." %(self.channel, message['data']))
46	 
47	 
48	if __name__ == "__main__":
49	    MySlave().start()
50	    time.sleep(10000)

说明

MySlave类 – slave节点主程序，用来启动MyJobWorkerThread的线程

MyJobWorkerThread类 – 从channel里获取派发的作业并将运行结果发送回master

测试

首先运行MySlave来定义派发作业channel。

然后运行MyMaster派发作业并显示执行结果。