python基于mysql实现的简单队列以及跨进程锁实例详解

yipeiwu_com6年前Python基础

通常在我们进行多进程应用开发的过程中,不可避免的会遇到多个进程访问同一个资源(临界资源)的状况,这时候必须通过加一个全局性的锁,来实现资源的同步访问(即:同一时间里只能有一个进程访问资源)。

举个例子如下:

假设我们用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]#

相关文章

在CentOS6上安装Python2.7的解决方法

在CentOS6上yum安装工具是基于Python2.6.6的,所以在CentOS6上默认安装的是Python2.6.6,因为要在服务器系统为CentOS6上部署生产环境,但是代码都是基...

Pytorch 实现计算分类器准确率(总分类及子分类)

分类器平均准确率计算: correct = torch.zeros(1).squeeze().cuda() total = torch.zeros(1).squeeze().cuda...

python django框架中使用FastDFS分布式文件系统的安装方法

python django框架中使用FastDFS分布式文件系统的安装方法

一、安装FastDFS 1-1:执行docker命令安装 # 安装tracker docker run -dti --network=host --name tracker -v /...

python实现年会抽奖程序

用python来实现一个抽奖程序,供大家参考,具体内容如下 主要功能有 1.从一个csv文件中读入所有员工工号 2.将这些工号初始到一个列表中 3.用random模块下的choice...

python pandas 如何替换某列的一个值

python pandas 如何替换某列的一个值

摘要:本文主要是讲解怎么样替换某一列的一个值。 应用场景: 假如我们有以下的数据集: 我们想把里面不是pre的字符串全部换成Nonpre,我们要怎么做呢? 做法很简单。 df['c...