Python使用分布式锁的代码演示示例

在计算机并发领域编程中总是会与锁打交道,锁又有很多种,互斥锁、自旋锁等等。

锁总是伴随着线程、进程这样的词汇出现,阮一峰有 一篇文章 对这些名词进行了简单易懂的解释。

我的理解是,使用线程、进程是为了实现并发从而获得性能的提升(利用多核CPU,多台服务器),但这种并发由于调度的不确定性,很容易出乱子,为了(在一些共享资源、关键节点上)不出乱子,又需要对资源加锁,在操作这个资源时控制这种并发,将乱子消灭。

很多语言都提供了一些线程级别的锁实现以及一些相应的工具,但在进程方面就无能为力了。而一个服务部署到生产环境,往往会部署多个实例,这种情况下,就经常会用到给不同进程用的锁,分布式锁便是在分布式系统中对某共享资源进行加锁的构件。

现在来试着展示一下在Python项目中如何使用简单的分布式互斥锁。

不使用分布式锁会怎样

先用一个简单的实例来演示一下,不使用分布式锁会出怎样的乱子。

假设商城系统要做秒杀活动,在redis中记录着 count:1 的信息,到秒杀时间点的时候,会收到许多的请求,这时各应用程序去查redis中count的值,若count还大于0,则将count-1,这样其他请求就不再能秒杀到了。

# -*- coding: utf-8 -*-
import os
import arrow
import redis
from multiprocessing import Pool

HOT_KEY = 'count'
r = redis.Redis(host='localhost', port=6379)

def seckilling():
  name = os.getpid()
  v = r.get(HOT_KEY)
  if int(v) > 0:
    print name, ' decr redis.'
    r.decr(HOT_KEY)
  else:
    print name, ' can not set redis.', v

def run_without_lock(name):
  while True:
    if arrow.now().second % 5 == 0:
      seckilling()
      return

if __name__ == '__main__':
  p = Pool(16)
  r.set(HOT_KEY, 1)
  for i in range(16):
    p.apply_async(run_without_lock, args=(i, ))
  print 'now 16 processes are going to get lock!'
  p.close()
  p.join()
  print('All subprocesses done.')

以上代码使用多进程来模仿这种并发请求场景,程序开始的时候将count设为1,之后各进程开始进入等待,当秒数为5的时候,所有进程同时去访问秒杀函数,来看一下效果:

Python使用分布式锁的代码演示示例

运行结果

Python使用分布式锁的代码演示示例

redis查询展示

从程序打印与查redis的结果来说并未如愿,本来秒杀商品只有一件,但却被成功抢购到了4次。这是由于各进程在get count的值时,对redis值更新的指令已经发出而还未进行完毕,会让其他进程认为自己可以购得。

这种问题可归为 不可重复读 种类的数据并发问题。

在这种毫无保护的情况下,其他常见并发问题幻读、脏读、第一第二类丢失更新等都有可能发生,这里不再一一举例。

使用ZooKeeper作分布式锁

作为致力于解决分布式协同问题的知名工具,利用zookeeper提供的API和它对于节点唯一性与顺序一致性的保证可以实现分式式锁。

实现思路为,各进程去创建 /exclusive_lock/lock 的结点,zookeeper保证只有一个client可以创建成功,那么便认为创建成功的那个client获得了锁,当它处理完业务后,将该node删除,其他client会监听到这个事件,并再次尝试创建该节点,如此进行下去。

Kazoo 库实现了这种Lock,使用起来非常简单,编程人员可以不用再去自己实现acquire,release等锁的通用接口。

同时在Python中,对锁的使用往往可以通过优雅的上下文管理器with。

def run_with_zk_lock(name):
  zk = KazooClient()
  zk.start()
  lock = zk.Lock("/lockpath", "my-identifier")
  while True:
    if arrow.now().second % 5 == 0:
      with lock:
        seckilling()
        return

Python使用分布式锁的代码演示示例

使用zk结果

Python使用分布式锁的代码演示示例

redis查询展示

当秒杀发生时,只有获得锁的进程可以去进行秒杀操作。

在锁的帮助下,程序按照预想的方式运行了。

使用redis作分布式锁

在redis的网站有一篇文章 专门介绍如何使用redis作为分布式锁,文尾还附带了对此文章的反对文章以及再次回击的文章,有点精彩。

文章提到了一个redlock的分布式锁设计。

设置锁的redis命令为 SET resource_name my_random_value NX PX 30000 ,当加NX参数时,若resouce_name不存在才会创建,若不存在则会向client返回不同的结果,利用这个机制,便只有一个client可以set成功,就像上面的zk一样了。

但是,实现这样一个分布式锁远不止这么简单,redis并不像zk一样是一个分布式协同工具,会向client做出分布式中各种一致性及容错、可用性的保证。

redis本身也是集群部署的,它们之间有着异步复制时间差、容错等问题可能会出现,要真正做到这个锁的实现在线上大规模分布式系统中可用,真的是要考虑各种情况,很不容易。

关于如何在语言上实现一个锁的接口,redlock的原理与代码实现,以及上述kazoo包里实现lock的源码,我会在另一篇专门的文章中说一下。

redlock-py 包是python语言中对上述文章的实现,我们现在使用它来进行尝试。

rlock = RedLock([{"host": "localhost", "port": 6379, "db": 0}, ])

def run_with_redis_lock(name):
  while True:
    if arrow.now().second % 5 == 0:
      with rlock:
        seckilling()
        return

Python使用分布式锁的代码演示示例

redis锁运行结果

运行结果和上面使用zk一样,符合程序设计预期。

以上只是基于python语言的一些代码展示,通过使用两个第三方包,来使用分布式锁来避免并发程序中混乱的产生。

但其实这中间是有一个断层的,即,这两个工具都是提供了一个机制,而并不是直接对外提供了操作锁的API,那么如何利用这个机制来实现这样的锁正是这两个第三方做的事情。

简单看过它们实现的源码,以及threading中一些lock的代码,发现在锁的实现上是有着共通之处的,都有通用的acquire与release方法,然后将 enter 与 exit 指向前面两个方法来实现上下文管理器with的用法。

此外,还可以利用关系型数据库如MySQL固有的锁机制来作为分布式锁,但由于数据库往往是系统的瓶颈所在,没有必要为它引入不必要的压力。同时,MySQL中的锁、隔离级别也有一大堆可说的,在github上找了一下也并未找到一个成熟的像上面的基于MySQL实现的对外暴露锁通用API的第三方包,故未能在上面加以展示。

想要说清楚这个事情并没有那么容易,之后我会尝试搞清楚如何写一个比较地道的锁,并对上面两个第三方包的具体实现加以研究,争取把这个断层补上。之后,或许可以尝试实现一下基于MySQL的类似第三方包,这需要对MySQL的一些机制搞得更加清楚才行。

相关推荐