对于运行在同一台服务器上的两个进程，可以通过加锁实现互斥执行，而对于运行在多个服务器上的任务仍然可以通过用加锁实现互斥，不过这个锁是分布式锁。这个分布式锁并没有那么神秘，实际上只要一个提供原子性的数据库即可。比如，在数据库的locks表里有一个记录（lock record），包含属性：
name：锁的名字，互斥的任务需要用名字相同的锁。一个持有锁的服务器通过不断的发送心跳，来更新这个记录，心跳的内容就是持有锁的时间戳（update_time），以及本机ip。也就是说，通过发送心跳来保证当前的服务器是活跃的，而其他服务器通过lock record中的update_time来判断当前活跃的服务器是否超时，一旦超时，其他的服务器就会去争夺锁，接管任务的执行，并发送心跳更新active_ip。
active_ip：持有锁的服务器的ip。
update_time：上次持有锁的时间，其他非活跃的服务器通过这个属性判断活跃的服务器是否超时，如果超时，则会争夺锁。
通过上面描述，这个框架中最重要的两个概念就是分布式锁和心跳。下面看一下分布式定时任务框架中是如何实现这两点的。当然，这个框架依赖于APScheduler，所以必须安装这个模块，具体APScheduler的介绍见我的上一篇文章：python第三方库系列之十三--定时任务apscheduler库，因为依赖APScheduler，所以这个框架很简单，只有一个类：
import datetime
import socket
import struct
import fcntl
from apscheduler.scheduler import Scheduler
def get_ip(ifname):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
return socket.inet_ntoa(fcntl.ioctl(
s.fileno(),
0x8915,
struct.pack('256s', ifname[:15])
)[20:24])
class MutexScheduler(Scheduler):
def __init__(self, local_ip, gconfig={}, **options):
Scheduler.__init__(self, gconfig, **options)
#self.ip = get_ip(settings.NETWORK_INTERFACE)
self.ip = local_ip
def mutex(self, lock=None, heartbeat=None, lock_else=None,
unactive_interval=datetime.timedelta(seconds=10)):
def mutex_func_gen(func):
def mtx_func():
if lock:
lock_rec = lock()
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
now = datetime.datetime.strptime(now, "%Y-%m-%d %H:%M:%S")
# execute mutex job when the server is active, or the other server is timeout.
if not lock_rec or lock_rec['active_ip'] == self.ip or (
lock_rec['update_time'] and now - lock_rec['update_time'] >= unactive_interval):
if lock_rec:
del lock_rec['active_ip']
del lock_rec['update_time']
if not lock_rec:
lock_rec = {}
lock_attrs = func(**lock_rec)
if not lock_attrs:
lock_attrs = {}
# send heart beat
heartbeat(self.ip, now, **lock_attrs)
else:
lock_else(lock_rec)
else:
func()
return mtx_func
self.mtx_func_gen = mutex_func_gen
def inner(func):
return func
return inner
def cron_schedule(self, **options):
def inner(func):
if hasattr(self, 'mtx_func_gen'):
func = self.mtx_func_gen(func)
func.job = self.add_cron_job(func, **options)
return func
return inner
mutex方法是核心，通过装饰器的方式提供互斥功能。在使用时：
@sched.mutex(lock = my_lock, heartbeat = my_heartbeat)job的原型是job(**attrs)，attrs就是存放在锁记录中的用户自定义属性，job可以有dict类型的返回值，这个返回值会存入锁记录中。
@sched.cron_schedule(second = '*')
def my_job(**attrs):
print 'my_job ticks'
#mutex装饰器必须用在cron_schedule装饰器之前，mutex主要是组装job。mutex的参数有：
#lock：函数，用于获取锁记录（lock record），函数原型：lock()。lock的返回值时dict，就是锁记录内容。
#heartbeat：函数，用于发出心跳，函数原型：heartbeat(ip, now, **attrs)。ip是本机ip；now是当前时间戳；attrs是一个dict，用于在锁记录中存放一些其他用户自定义信息。
#lock_else：函数，在没有获得锁时执行，函数原型：lock_else(lock_rec)。lock_rec是锁记录，包含active_ip，update_time以及用户自定义的属性。
#unactive_interval：datetime.timedelta类型，超时时间，也就是说当前时间减去update_time大于unactive_interval的话，就代表超时。类中默认值unactive_interval=datetime.timedelta(seconds=10)是默认10s。
#在使用这个类时，必须实现自己的lock，heartbeat以及lock_else函数。
下面，看一下具体使用的例子，使用的mongodb存放分布式锁。
import apscheduler.events这个任务很简单就是定时打印整数序列。同时在两台服务器上部署运行，可以发现只有一台服务器会输出整数序列。
import datetime
import time
import pymongo
import sys
import mtxscheduler
sched = mtxscheduler.MutexScheduler()
mongo = pymongo.Connection(host = '127.0.0.1', port = 27017)
lock_store = mongo['lockstore']['locks']
def lock():
conn = connect_adms_db()
lock_name = 'xxx'
sql = "select name, active_ip, update_time from locks where name='%s';" % lock_name
log.info("sql:%s" % sql)
res = conn.execute(sql)[0]
conn.close()
tuple = {'name': res["name"], 'active_ip': res["active_ip"], 'update_time': res["update_time"]}
return tuple
def hb(ip, now, **attrs):
attrs['active_ip'] = ip
attrs['update_time'] = now
conn = connect_adms_db()
lock_name = 'xxx'
sql = "update locks set active_ip='%(ip)s', update_time='%(update_time)s' " \
"where name='%(name)s'" % {'ip': ip, 'update_time': now, 'name': lock_name}
log.info("sql:%s" % sql)
res = conn.execute(sql)
conn.close()
def le(lock_rec):
if lock_rec:
print 'active ip', lock_rec['active_ip']
else:
print 'lock else'
i = 0
@sched.mutex(lock = lock, heartbeat = hb, lock_else = le)
@sched.cron_schedule(second = '*')
def job(**attr):
global i
i += 1
print i
def err_listener(ev):
if ev.exception:
print sys.exc_info()
sched.add_listener(err_listener, apscheduler.events.EVENT_JOB_ERROR)
sched.start()
time.sleep(10)
还有使用redis，mongodb存储锁。