python基于mysql实现的简单队列以及跨进程锁实例详解
通常在我们进行多进程应用开发的过程中,不可避免的会遇到多个进程访问同一个资源(临界资源)的状况,这时候必须通过加一个全局性的锁,来实现资源的同步访问(即:同一时间里只能有一个进程访问资源)。
举个例子如下:
假设我们用mysql来实现一个任务队列,实现的过程如下:
1. 在mysql中创建job表,用于储存队列任务,如下:
create table jobs( id auto_increment not null primary key, message text not null, job_status not null default 0 );
message 用来存储任务信息,job_status用来标识任务状态,假设只有两种状态,0:在队列中, 1:已出队列
2. 有一个生产者进程,往job表中放新的数据,进行排队:
insert into jobs(message) values('msg1');
3.假设有多个消费者进程,从job表中取排队信息,要做的操作如下:
select * from jobs where job_status=0 order by id asc limit 1; update jobs set job_status=1 where id = ?; -- id为刚刚取得的记录id
4. 如果没有跨进程的锁,两个消费者进程有可能同时取到重复的消息,导致一个消息被消费多次。这种情况是我们不希望看到的,于是,我们需要实现一个跨进程的锁。
=========================分割线=======================================
说到跨进程的锁实现,我们主要有几种实现方式:
(1)信号量
(2)文件锁fcntl
(3)socket(端口号绑定)
(4)signal
这几种方式各有利弊,总体来说前2种方式可能多一点,这里我就不详细说了,大家可以去查阅资料。
查资料的时候发现mysql中有锁的实现,适用于对于性能要求不是很高的应用场景,大并发的分布式访问可能会有瓶颈.
对此用python实现了一个demo,如下:
文件名:glock.py
#!/usr/bin/env python2.7 # # -*- coding:utf-8 -*- # # desc : # import logging, time import mysqldb class glock: def __init__(self, db): self.db = db def _execute(self, sql): cursor = self.db.cursor() try: ret = none cursor.execute(sql) if cursor.rowcount != 1: logging.error("multiple rows returned in mysql lock function.") ret = none else: ret = cursor.fetchone() cursor.close() return ret except exception, ex: logging.error("execute sql \"%s\" failed! exception: %s", sql, str(ex)) cursor.close() return none def lock(self, lockstr, timeout): sql = "select get_lock('%s', %s)" % (lockstr, timeout) ret = self._execute(sql) if ret[0] == 0: logging.debug("another client has previously locked '%s'.", lockstr) return false elif ret[0] == 1: logging.debug("the lock '%s' was obtained successfully.", lockstr) return true else: logging.error("error occurred!") return none def unlock(self, lockstr): sql = "select release_lock('%s')" % (lockstr) ret = self._execute(sql) if ret[0] == 0: logging.debug("the lock '%s' the lock is not released(the lock was not established by this thread).", lockstr) return false elif ret[0] == 1: logging.debug("the lock '%s' the lock was released.", lockstr) return true else: logging.error("the lock '%s' did not exist.", lockstr) return none #init logging def init_logging(): sh = logging.streamhandler() logger = logging.getlogger() logger.setlevel(logging.debug) formatter = logging.formatter('%(asctime)s -%(module)s:%(filename)s-l%(lineno)d-%(levelname)s: %(message)s') sh.setformatter(formatter) logger.addhandler(sh) logging.info("current log level is : %s",logging.getlevelname(logger.geteffectivelevel())) def main(): init_logging() db = mysqldb.connect(host='localhost', user='root', passwd='') lock_name = 'queue' l = glock(db) ret = l.lock(lock_name, 10) if ret != true: logging.error("can't get lock! exit!") quit() time.sleep(10) logging.info("you can do some synchronization work across processes!") ##todo ## you can do something in here ## l.unlock(lock_name) if __name__ == "__main__": main()
在main函数里:
l.lock(lock_name, 10) 中,10是表示timeout的时间是10秒,如果10秒还获取不了锁,就返回,执行后面的操作。
在这个demo中,在标记todo的地方,可以将消费者从job表中取消息的逻辑放在这里。即分割线以上的.
2.假设有多个消费者进程,从job表中取排队信息,要做的操作如下:
select * from jobs where job_status=0 order by id asc limit 1; update jobs set job_status=1 where id = ?; -- id为刚刚取得的记录id
这样,就能保证多个进程访问临界资源时同步进行了,保证数据的一致性。
测试的时候,启动两个glock.py, 结果如下:
[@tj-10-47 test]# ./glock.py 2014-03-14 17:08:40,277 -glock:glock.py-l70-info: current log level is : debug 2014-03-14 17:08:40,299 -glock:glock.py-l43-debug: the lock 'queue' was obtained successfully. 2014-03-14 17:08:50,299 -glock:glock.py-l81-info: you can do some synchronization work across processes! 2014-03-14 17:08:50,299 -glock:glock.py-l56-debug: the lock 'queue' the lock was released.
可以看到第一个glock.py是 17:08:50解锁的,下面的glock.py是在17:08:50获取锁的,可以证实这样是完全可行的。
[@tj-10-47 test]# ./glock.py 2014-03-14 17:08:46,873 -glock:glock.py-l70-info: current log level is : debug 2014-03-14 17:08:50,299 -glock:glock.py-l43-debug: the lock 'queue' was obtained successfully. 2014-03-14 17:09:00,299 -glock:glock.py-l81-info: you can do some synchronization work across processes! 2014-03-14 17:09:00,300 -glock:glock.py-l56-debug: the lock 'queue' the lock was released. [@tj-10-47 test]#
下一篇: 学车这事还是先缓一缓