- 浏览: 25990 次
- 性别:
- 来自: 北京
最新评论
-
明兜3号:
基于spring+quartz的分布式任务调度网盘地址:htt ...
Quartz集成springMVC (持久化任务、集群和分布式)
分布式锁1 Java常用技术方案
前言:
由于在平时的工作中,线上服务器是分布式多台部署的,经常会面临解决分布式场景下数据一致性的问题,那么就要利用分布式锁来解决这些问题。所以自己结合实际工作中的一些经验和网上看到的一些资料,做一个讲解和总结。希望这篇文章可以方便自己以后查阅,同时要是能帮助到他人那也是很好的。
===============================================================长长的分割线====================================================================
正文:
第一步,自身的业务场景:
在我日常做的项目中,目前涉及了以下这些业务场景:
场景一: 比如分配任务场景。在这个场景中,由于是公司的业务后台系统,主要是用于审核人员的审核工作,并发量并不是很高,而且任务的分配规则设计成了通过审核人员每次主动的请求拉取,然后服务端从任务池中随机的选取任务进行分配。这个场景看到这里你会觉得比较单一,但是实际的分配过程中,由于涉及到了按用户聚类的问题,所以要比我描述的复杂,但是这里为了说明问题,大家可以把问题简单化理解。那么在使用过程中,主要是为了避免同一个任务同时被两个审核人员获取到的问题。我最终使用了基于数据库资源表的分布式锁来解决的问题。
场景二: 比如支付场景。在这个场景中,我提供给用户三个用于保护用户隐私的手机号码(这些号码是从运营商处获取的,和真实手机号码看起来是一样的),让用户选择其中一个进行购买,用户购买付款后,我需要将用户选择的号码分配给用户使用,同时也要将没有选择的释放掉。在这个过程中,给用户筛选的号码要在一定时间内(用户筛选正常时间范围内)让当前用户对这个产品具有独占性,以便保证付款后是100%可以拿到;同时由于产品资源池的资源有限,还要保持资源的流动性,即不能让资源长时间被某个用户占用着。对于服务的设计目标,一期项目上线的时候至少能够支持峰值qps为300的请求,同时在设计的过程中要考虑到用户体验的问题。我最终使用了memecahed的add()方法和基于数据库资源表的分布式锁来解决的问题。
场景三: 我有一个数据服务,每天调用量在3亿,每天按86400秒计算的qps在4000左右,由于服务的白天调用量要明显高于晚上,所以白天下午的峰值qps达到6000的,一共有4台服务器,单台qps要能达到3000以上。我最终使用了redis的setnx()和expire()的分布式锁解决的问题。
场景四:场景一和场景二的升级版。在这个场景中,不涉及支付。但是由于资源分配一次过程中,需要保持涉及一致性的地方增加,而且一期的设计目标要达到峰值qps500,所以需要我们对场景进一步的优化。我最终使用了redis的setnx()、expire()和基于数据库表的分布式锁来解决的问题。
看到这里,不管你觉得我提出的业务场景qps是否足够大,都希望你能继续看下去,因为无论你身处一个什么样的公司,最开始的工作可能都需要从最简单的做起。不要提阿里和腾讯的业务场景qps如何大,因为在这样的大场景中你未必能亲自参与项目,亲自参与项目未必能是核心的设计者,是核心的设计者未必能独自设计。如果能真能满足以上三条,关闭页面可以不看啦,如果不是的话,建议还是看完,我有说的不足的地方欢迎提出建议,我说的好的地方,也希望给我点个赞或者评论一下,算是对我最大的鼓励哈。
第二步,分布式锁的解决方式:
1. 首先明确一点,有人可能会问是否可以考虑采用ReentrantLock来实现,但是实际上去实现的时候是有问题的,ReentrantLock的lock和unlock要求必须是在同一线程进行,而分布式应用中,lock和unlock是两次不相关的请求,因此肯定不是同一线程,因此导致无法使用ReentrantLock。
2. 基于数据库表做乐观锁,用于分布式锁。
3. 使用memcached的add()方法,用于分布式锁。
4. 使用memcached的cas()方法,用于分布式锁。(不常用)
5. 使用redis的setnx()、expire()方法,用于分布式锁。
6. 使用redis的setnx()、get()、getset()方法,用于分布式锁。
7. 使用redis的watch、multi、exec命令,用于分布式锁。(不常用)
8. 使用zookeeper,用于分布式锁。(不常用)
第三步,基于数据库资源表做乐观锁,用于分布式锁:
1. 首先说明乐观锁的含义:
大多数是基于数据版本(version)的记录机制实现的。何谓数据版本号?即为数据增加一个版本标识,在基于数据库表的版本解决方案中,一般是通过为数据库表添加一个 “version”字段来实现读取出数据时,将此版本号一同读出,之后更新时,对此版本号加1。
在更新过程中,会对版本号进行比较,如果是一致的,没有发生改变,则会成功执行本次操作;如果版本号不一致,则会更新失败。
2. 对乐观锁的含义有了一定的了解后,结合具体的例子,我们来推演下我们应该怎么处理:
(1). 假设我们有一张资源表,如下图所示: t_resource , 其中有6个字段id, resoource, state, add_time, update_time, version,分别表示表主键、资源、分配状态(1未分配 2已分配)、资源创建时间、资源更新时间、资源数据版本号。
(4). 假设我们现在我们对id=5780这条数据进行分配,那么非分布式场景的情况下,我们一般先查询出来state=1(未分配)的数据,然后从其中选取一条数据可以通过以下语句进行,如果可以更新成功,那么就说明已经占用了这个资源
update t_resource set state=2 where state=1 and id=5780。
(5). 如果在分布式场景中,由于数据库的update操作是原子是原子的,其实上边这条语句理论上也没有问题,但是这条语句如果在典型的“ABA”情况下,我们是无法感知的。有人可能会问什么是“ABA”问题呢?大家可以网上搜索一下,这里我说简单一点就是,如果在你第一次select和第二次update过程中,由于两次操作是非原子的,所以这过程中,如果有一个线程,先是占用了资源(state=2),然后又释放了资源(state=1),实际上最后你执行update操作的时候,是无法知道这个资源发生过变化的。也许你会说这个在你说的场景中应该也还好吧,但是在实际的使用过程中,比如银行账户存款或者扣款的过程中,这种情况是比较恐怖的。
(6). 那么如果使用乐观锁我们如何解决上边的问题呢?
a. 先执行select操作查询当前数据的数据版本号,比如当前数据版本号是26:
select id, resource, state,version from t_resource where state=1 and id=5780;
b. 执行更新操作:
update t_resoure set state=2, version=27, update_time=now() where resource=xxxxxx and state=1 and version=26
c. 如果上述update语句真正更新影响到了一行数据,那就说明占位成功。如果没有更新影响到一行数据,则说明这个资源已经被别人占位了。
3. 通过2中的讲解,相信大家已经对如何基于数据库表做乐观锁有有了一定的了解了,但是这里还是需要说明一下基于数据库表做乐观锁的一些缺点:
(1). 这种操作方式,使原本一次的update操作,必须变为2次操作: select版本号一次;update一次。增加了数据库操作的次数。
(2). 如果业务场景中的一次业务流程中,多个资源都需要用保证数据一致性,那么如果全部使用基于数据库资源表的乐观锁,就要让每个资源都有一张资源表,这个在实际使用场景中肯定是无法满足的。而且这些都基于数据库操作,在高并发的要求下,对数据库连接的开销一定是无法忍受的。
(3). 乐观锁机制往往基于系统中的数据存储逻辑,因此可能会造成脏数据被更新到数据库中。在系统设计阶段,我们应该充分考虑到这些情况出现的可能性,并进行相应调整,如将乐观锁策略在数据库存储过程中实现,对外只开放基于此存储过程的数据更新途径,而不是将数据库表直接对外公开。
4. 讲了乐观锁的实现方式和缺点,是不是会觉得不敢使用乐观锁了呢???当然不是,在文章开头我自己的业务场景中,场景1和场景2的一部分都使用了基于数据库资源表的乐观锁,已经很好的解决了线上问题。所以大家要根据的具体业务场景选择技术方案,并不是随便找一个足够复杂、足够新潮的技术方案来解决业务问题就是好方案?!比如,如果在我的场景一中,我使用zookeeper做锁,可以这么做,但是真的有必要吗???答案觉得是没有必要的!!!
第四步,使用memcached的add()方法,用于分布式锁:
对于使用memcached的add()方法做分布式锁,这个在互联网公司是一种比较常见的方式,而且基本上可以解决自己手头上的大部分应用场景。在使用这个方法之前,只要能搞明白memcached的add()和set()的区别,并且知道为什么能用add()方法做分布式锁就好。如果还不知道add()和set()方法,请直接百度吧,这个需要自己了解一下。
我在这里想说明的是另外一个问题,人们在关注分布式锁设计的好坏时,还会重点关注这样一个问题,那就是是否可以避免死锁问题???!!!
如果使用memcached的add()命令对资源占位成功了,那么是不是就完事儿了呢?当然不是!我们需要在add()的使用指定当前添加的这个key的有效时间,如果不指定有效时间,正常情况下,你可以在执行完自己的业务后,使用delete方法将这个key删除掉,也就是释放了占用的资源。但是,如果在占位成功后,memecached或者自己的业务服务器发生宕机了,那么这个资源将无法得到释放。所以通过对key设置超时时间,即便发生了宕机的情况,也不会将资源一直占用,可以避免死锁的问题。
第五步,使用memcached的cas()方法,用于分布式锁:
下篇文章我们再细说!
第六步,使用redis的setnx()、expire()方法,用于分布式锁:
对于使用redis的setnx()、expire()来实现分布式锁,这个方案相对于memcached()的add()方案,redis占优势的是,其支持的数据类型更多,而memcached只支持String一种数据类型。除此之外,无论是从性能上来说,还是操作方便性来说,其实都没有太多的差异,完全看你的选择,比如公司中用哪个比较多,你就可以用哪个。
首先说明一下setnx()命令,setnx的含义就是SET if Not Exists,其主要有两个参数 setnx(key, value)。该方法是原子的,如果key不存在,则设置当前key成功,返回1;如果当前key已经存在,则设置当前key失败,返回0。但是要注意的是setnx命令不能设置key的超时时间,只能通过expire()来对key设置。
具体的使用步骤如下:
1. setnx(lockkey, 1) 如果返回0,则说明占位失败;如果返回1,则说明占位成功
2. expire()命令对lockkey设置超时时间,为的是避免死锁问题。
3. 执行完业务代码后,可以通过delete命令删除key。
这个方案其实是可以解决日常工作中的需求的,但从技术方案的探讨上来说,可能还有一些可以完善的地方。比如,如果在第一步setnx执行成功后,在expire()命令执行成功前,发生了宕机的现象,那么就依然会出现死锁的问题,所以如果要对其进行完善的话,可以使用redis的setnx()、get()和getset()方法来实现分布式锁。
第七步,使用redis的setnx()、get()、getset()方法,用于分布式锁:
这个方案的背景主要是在setnx()和expire()的方案上针对可能存在的死锁问题,做了一版优化。
那么先说明一下这三个命令,对于setnx()和get()这两个命令,相信不用再多说什么。那么getset()命令?这个命令主要有两个参数 getset(key,newValue)。该方法是原子的,对key设置newValue这个值,并且返回key原来的旧值。假设key原来是不存在的,那么多次执行这个命令,会出现下边的效果:
1. getset(key, "value1") 返回nil 此时key的值会被设置为value1
2. getset(key, "value2") 返回value1 此时key的值会被设置为value2
3. 依次类推!
介绍完要使用的命令后,具体的使用步骤如下:
1. setnx(lockkey, 当前时间+过期超时时间) ,如果返回1,则获取锁成功;如果返回0则没有获取到锁,转向2。
2. get(lockkey)获取值oldExpireTime ,并将这个value值与当前的系统时间进行比较,如果小于当前系统时间,则认为这个锁已经超时,可以允许别的请求重新获取,转向3。
3. 计算newExpireTime=当前时间+过期超时时间,然后getset(lockkey, newExpireTime) 会返回当前lockkey的值currentExpireTime。
4. 判断currentExpireTime与oldExpireTime 是否相等,如果相等,说明当前getset设置成功,获取到了锁。如果不相等,说明这个锁又被别的请求获取走了,那么当前请求可以直接返回失败,或者继续重试。
5. 在获取到锁之后,当前线程可以开始自己的业务处理,当处理完毕后,比较自己的处理时间和对于锁设置的超时时间,如果小于锁设置的超时时间,则直接执行delete释放锁;如果大于锁设置的超时时间,则不需要再锁进行处理。
注意: 这个方案我当初在线上使用的时候是没有问题的,所以当初写这篇文章时也认为是没有问题的。但是截止到2017.05.13(周六),自己在重新回顾这篇文章时,看了文章下网友的很多评论,我发现有两个问题比较集中:
问题1: 在“get(lockkey)获取值oldExpireTime ”这个操作与“getset(lockkey, newExpireTime) ”这个操作之间,如果有N个线程在get操作获取到相同的oldExpireTime后,然后都去getset,会不会返回的newExpireTime都是一样的,都会是成功,进而都获取到锁???
我认为这套方案是不存在这个问题的。依据有两条: 第一,redis是单进程单线程模式,串行执行命令。 第二,在串行执行的前提条件下,getset之后会比较返回的currentExpireTime与oldExpireTime 是否相等。
问题2: 在“get(lockkey)获取值oldExpireTime ”这个操作与“getset(lockkey, newExpireTime) ”这个操作之间,如果有N个线程在get操作获取到相同的oldExpireTime后,然后都去getset,假设第1个线程获取锁成功,其他锁获取失败,但是获取锁失败的线程它发起的getset命令确实执行了,这样会不会造成第一个获取锁的线程设置的锁超时时间一直在延长???
我认为这套方案确实存在这个问题的可能。但我个人认为这个微笑的误差是可以忽略的,不过技术方案上存在缺陷,大家可以自行抉择哈。
第八步,使用redis的watch、multi、exec命令,用于分布式锁:
下篇文章我们再细说!
第九步,使用zookeeper,用于分布式锁:
下篇文章我们再细说!
第十步,总结:
综上,关于分布式锁的第一篇文章我就写到这儿了,在文章中主要说明了日常项目中会比较常用到四种方案,大家掌握了这四种方案,其实在日常的工作中就可以解决很多业务场景下的分布式锁的问题。从文章开头我自己的实际使用中,也可以看到,这么说完全是有一定的依据。对于另外那三种方案,我会在下一篇关于分布式锁的文章中,和大家再探讨一下。
常用的四种方案:
1. 基于数据库表做乐观锁,用于分布式锁。
2. 使用memcached的add()方法,用于分布式锁。
3. 使用redis的setnx()、expire()方法,用于分布式锁。
4. 使用redis的setnx()、get()、getset()方法,用于分布式锁。
不常用但是可以用于技术方案探讨的:
1. 使用memcached的cas()方法,用于分布式锁。
2. 使用redis的watch、multi、exec命令,用于分布式锁。
3. 使用zookeeper,用于分布式锁。
转载请注明来自博客园http://www.cnblogs.com/PurpleDream/p/5559352.html ,版权归本人和博客园所有,谢谢!
分布式锁2 Java非常用技术方案探讨之ZooKeeper
前言:
由于在平时的工作中,线上服务器是分布式多台部署的,经常会面临解决分布式场景下数据一致性的问题,那么就要利用分布式锁来解决这些问题。以自己结合实际工作中的一些经验和网上看到的一些资料,做一个讲解和总结。之前我已经写了一篇关于分布式锁的文章: 分布式锁1 Java常用技术方案 。上一篇文章中主要写的是在日常项目中,较为常见的几种实现分布式锁的方法。通过这些方法,基本上可以解决我们日常工作中大部分场景下使用分布式锁的问题。
本篇文章主要是在上一篇文章的基础上,介绍一些虽然日常工作中不常用或者比较实现起来比较重,但是可以作为技术方案学习了解一下的分布式锁方案。希望这篇文章可以方便自己以后查阅,同时要是能帮助到他人那也是很好的。
===============================================================长长的分割线====================================================================
正文:
第一步,使用zookeeper节点名称唯一性,用于分布式锁:
关于zookeeper集群的搭建,可以参考我之前写的一篇文章: ZooKeeper1 利用虚拟机搭建自己的ZooKeeper集群
zookeeper抽象出来的节点结构是一个和文件系统类似的小型的树状的目录结构,同时zookeeper机制规定:同一个目录下只能有一个唯一的文件名。例如:我们在zookeeper的根目录下,由两个客户端同时创建一个名为/myDistributeLock,只有一个客户端可以成功。
上述方案和memcached的add()方法、redis的setnx()方法实现分布式锁有着相同的思路。这样的方案实现起来如果不考虑搭建和维护zookeeper集群的成本,由于正确性和可靠性是zookeeper机制自己保证的,实现还是比较简单的。
第二步,使用zookeeper临时顺序节点,用于分布式锁:
在讨论这套方案之前,我们有必要先“吹毛求疵”般的说明一下使用zookeeper节点名称唯一性来做分布式锁这个方案的缺点。比如,当许多线程在等待一个锁时,如果锁得到释放的时候,那么所有客户端都被唤醒,但是仅仅有一个客户端得到锁。在这个过程中,大量的线程根本没有获得锁的可能性,但是也会引起大量的上下文切换,这个系统开销也是不小的,对于这样的现象有一个专业名词,称之为“惊群效应”。
我们首先说明一下zookeeper的顺序节点、临时节点和watcher机制:
所谓顺序节点,假如我们在/myDisLocks/目录下创建3个节点,zookeeper集群会按照发起创建的顺序来创建节点,节点分别为/myDisLocks/0000000001、/myDisLocks/0000000002、/myDisLocks/0000000003。
所谓临时节点,临时节点由某个客户端创建,当客户端与zookeeper集群断开连接,则该节点自动被删除。
所谓对于watcher机制,大家可以参考Apache ZooKeeper Watcher机制源码解释。当然如果你之前不知道watcher机制是个什么东东,不建议你直接去看前边我提供的文章链接,这样你极有可能忘掉我们的讨论主线,即分布式锁的实现方案,而陷入到watcher机制的源码实现中。所以你也可以先看看下面的具体方案,猜测一下watcher是用来干嘛的,我这里先总结一句话做个引子: 所谓watcher机制,你可以简单一点儿理解成任何一个连接zookeeper的客户端可以通过watcher机制关注自己感兴趣的节点的增删改查,当这个节点发生增删改查的操作时,会“广播”自己的消息,所有对此感兴趣的节点可以在收到这些消息后,根据自己的业务需要执行后续的操作。
具体的使用步骤如下:
1. 每个业务线程调用create()方法创建名为“/myDisLocks/thread”的节点,需要注意的是,这里节点的创建类型需要设置为EPHEMERAL_SEQUENTIAL,即节点类型为临时顺序节点。此时/myDisLocks节点下会出现诸如/myDisLocks/thread0000000001、/myDisLocks/thread0000000002、/myDisLocks/thread0000000003这样的子节点。
2. 每个业务线程调用getChildren(“myDisLocks”)方法来获取/myDisLocks这个节点下所有已经创建的子节点。
3. 每个业务线程获取到所有子节点的路径之后,如果发现自己在步骤1中创建的节点的尾缀编号是所有节点中序号最小的,那么就认为自己获得了锁。
4. 如果在步骤3中发现自己并非是所有子节点中序号最小的,说明自己还没有获取到锁。使用watcher机制监视比自己创建节点的序列号小的节点(比自己创建的节点小的最大节点),进入等待。比如,如果当前业务线程创建的节点是/myDisLocks/thread0000000003,那么在没有获取到锁的情况下,他只需要监视/myDisLocks/thread0000000002的情况。只有当/myDisLocks/thread0000000002获取到锁并释放之后,当前业务线程才启动获取锁,这样可以避免一个业务线程释放锁之后,其他所有线程都去竞争锁,引起不必要的上下文切换,最终造成“惊群现象”。
5. 释放锁的过程相对比较简单,就是删除自己创建的那个子节点即可。
注意: 这个方案实现的分布式锁还带着一点儿公平锁的味道!为什么呢?我们在利用每个节点的序号进行排队以此来避免进群现象时,实际上所有业务线程获得锁的顺序就是自己创建节点的顺序,也就是哪个业务线程先来,哪个就可以最快获得锁。
下面贴出我自己实现的上述方案的代码:
1. 代码中有两个Java类: MyDistributedLockByZK.java和LockWatcher.java。其中MyDistributedLockByZK.java中的main函数利用线程池启动5个线程,以此来模拟多个业务线程竞争锁的情况;而LockWatcher.java定义分布式锁和实现了watcher机制。
2. 同时,我使用的zookeeper集群是自己以前利用VMWare搭建的集群,所以zookeeper链接是192.168.224.170:2181,大家可以根据替换成自己的zookeeper链接即可。
复制代码
1 public class MyDistributedLockByZK {
2 /** 线程池 **/
3 private static ExecutorService executorService = null;
4 private static final int THREAD_NUM = 5;
5 private static int threadNo = 0;
6 private static CountDownLatch threadCompleteLatch = new CountDownLatch(THREAD_NUM);
7
8 /** ZK的相关配置常量 **/
9 private static final String CONNECTION_STRING = "192.168.224.170:2181";
10 private static final int SESSION_TIMEOUT = 10000;
11 // 此变量在LockWatcher中也有一个同名的静态变量,正式使用的时候,提取到常量类中共同维护即可。
12 private static final String LOCK_ROOT_PATH = "/myDisLocks";
13
14 public static void main(String[] args) {
15 // 定义线程池
16 executorService = Executors.newFixedThreadPool(THREAD_NUM, new ThreadFactory() {
17 @Override
18 public Thread newThread(Runnable r) {
19 String name = String.format("第[%s]个测试线程", ++threadNo);
20 Thread ret = new Thread(Thread.currentThread().getThreadGroup(), r, name, 0);
21 ret.setDaemon(false);
22 return ret;
23 }
24 });
25
26 // 启动线程
27 if (executorService != null) {
28 startProcess();
29 }
30 }
31
32 /**
33 * @author zhangyi03
34 * @date 2017-5-23 下午5:57:27
35 * @description 模拟并发执行任务
36 */
37 public static void startProcess() {
38 Runnable disposeBusinessRunnable= new Thread(new Runnable() {
39 public void run() {
40 String threadName = Thread.currentThread().getName();
41
42 LockWatcher lock = new LockWatcher(threadCompleteLatch);
43 try {
44 /** 步骤1: 当前线程创建ZK连接 **/
45 lock.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
46
47 /** 步骤2: 创建锁的根节点 **/
48 // 注意,此处创建根节点的方式其实完全可以在初始化的时候由主线程单独进行根节点的创建,没有必要在业务线程中创建。
49 // 这里这样写只是一种思路而已,不必局限于此
50 synchronized (MyDistributedLockByZK.class){
51 lock.createPersistentPath(LOCK_ROOT_PATH, "该节点由" + threadName + "创建", true);
52 }
53
54 /** 步骤3: 开启锁竞争并执行任务 **/
55 lock.getLock();
56 } catch (Exception e) {
57 e.printStackTrace();
58 }
59 }
60 });
61
62 for (int i = 0; i < THREAD_NUM; i++) {
63 executorService.execute(disposeBusinessRunnable);
64 }
65 executorService.shutdown();
66
67 try {
68 threadCompleteLatch.await();
69 System.out.println("所有线程运行结束!");
70 } catch (InterruptedException e) {
71 e.printStackTrace();
72 }
73 }
74 }
复制代码
复制代码
1 public class LockWatcher implements Watcher {
2 /** 成员变量 **/
3 private ZooKeeper zk = null;
4 // 当前业务线程竞争锁的时候创建的节点路径
5 private String selfPath = null;
6 // 当前业务线程竞争锁的时候创建节点的前置节点路径
7 private String waitPath = null;
8 // 确保连接zk成功;只有当收到Watcher的监听事件之后,才执行后续的操作,否则请求阻塞在createConnection()创建ZK连接的方法中
9 private CountDownLatch connectSuccessLatch = new CountDownLatch(1);
10 // 标识线程是否执行完任务
11 private CountDownLatch threadCompleteLatch = null;
12
13 /** ZK的相关配置常量 **/
14 private static final String LOCK_ROOT_PATH = "/myDisLocks";
15 private static final String LOCK_SUB_PATH = LOCK_ROOT_PATH + "/thread";
16
17 public LockWatcher(CountDownLatch latch) {
18 this.threadCompleteLatch = latch;
19 }
20
21 @Override
22 public void process(WatchedEvent event) {
23 if (event == null) {
24 return;
25 }
26
27 // 通知状态
28 Event.KeeperState keeperState = event.getState();
29 // 事件类型
30 Event.EventType eventType = event.getType();
31
32 // 根据通知状态分别处理
33 if (Event.KeeperState.SyncConnected == keeperState) {
34 if ( Event.EventType.None == eventType ) {
35 System.out.println(Thread.currentThread().getName() + "成功连接上ZK服务器");
36 // 此处代码的主要作用是用来辅助判断当前线程确实已经连接上ZK
37 connectSuccessLatch.countDown();
38 }else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
39 System.out.println(Thread.currentThread().getName() + "收到情报,排我前面的家伙已挂,我准备再次确认我是不是最小的节点!?");
40 try {
41 if(checkMinPath()){
42 getLockSuccess();
43 }
44 } catch (Exception e) {
45 e.printStackTrace();
46 }
47 }
48 } else if ( Event.KeeperState.Disconnected == keeperState ) {
49 System.out.println(Thread.currentThread().getName() + "与ZK服务器断开连接");
50 } else if ( Event.KeeperState.AuthFailed == keeperState ) {
51 System.out.println(Thread.currentThread().getName() + "权限检查失败");
52 } else if ( Event.KeeperState.Expired == keeperState ) {
53 System.out.println(Thread.currentThread().getName() + "会话失效");
54 }
55 }
56
57 /**
58 * @author zhangyi03
59 * @date 2017-5-23 下午6:07:03
60 * @description 创建ZK连接
61 * @param connectString ZK服务器地址列表
62 * @param sessionTimeout Session超时时间
63 * @throws IOException
64 * @throws InterruptedException
65 */
66 public void createConnection(String connectString, int sessionTimeout) throws IOException, InterruptedException {
67 zk = new ZooKeeper(connectString, sessionTimeout, this);
68 // connectSuccessLatch.await(1, TimeUnit.SECONDS) 正式实现的时候可以考虑此处是否采用超时阻塞
69 connectSuccessLatch.await();
70 }
71
72 /**
73 * @author zhangyi03
74 * @date 2017-5-23 下午6:15:48
75 * @description 创建ZK节点
76 * @param path 节点path
77 * @param data 初始数据内容
78 * @param needWatch
79 * @return
80 * @throws KeeperException
81 * @throws InterruptedException
82 */
83 public boolean createPersistentPath(String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
84 if(zk.exists(path, needWatch) == null){
85 String result = zk.create( path,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
86 System.out.println(Thread.currentThread().getName() + "创建节点成功, path: " + result + ", content: " + data);
87 }
88 return true;
89 }
90
91 /**
92 * @author zhangyi03
93 * @date 2017-5-23 下午6:24:46
94 * @description 获取分布式锁
95 * @throws KeeperException
96 * @throws InterruptedException
97 */
98 public void getLock() throws Exception {
99 selfPath = zk.create(LOCK_SUB_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
100 System.out.println(Thread.currentThread().getName() + "创建锁路径:" + selfPath);
101 if(checkMinPath()){
102 getLockSuccess();
103 }
104 }
105
106 /**
107 * @author zhangyi03
108 * @date 2017-5-23 下午7:02:41
109 * @description 获取锁成功
110 * @throws KeeperException
111 * @throws InterruptedException
112 */
113 private void getLockSuccess() throws KeeperException, InterruptedException {
114 if(zk.exists(selfPath, false) == null){
115 System.err.println(Thread.currentThread().getName() + "本节点已不在了...");
116 return;
117 }
118 System.out.println(Thread.currentThread().getName() + "获取锁成功,开始处理业务数据!");
119 Thread.sleep(2000);
120 System.out.println(Thread.currentThread().getName() + "处理业务数据完成,删除本节点:" + selfPath);
121 zk.delete(selfPath, -1);
122 releaseConnection();
123 threadCompleteLatch.countDown();
124 }
125
126 /**
127 * @author zhangyi03
128 * @date 2017-5-23 下午7:06:46
129 * @description 关闭ZK连接
130 */
131 private void releaseConnection() {
132 if (zk != null) {
133 try {
134 zk.close();
135 } catch (InterruptedException e) {
136 e.printStackTrace();
137 }
138 }
139 System.out.println(Thread.currentThread().getName() + "释放ZK连接");
140 }
141
142 /**
143 * @author zhangyi03
144 * @date 2017-5-23 下午6:57:14
145 * @description 检查自己是不是最小的节点
146 * @param selfPath
147 * @return
148 * @throws KeeperException
149 * @throws InterruptedException
150 */
151 private boolean checkMinPath() throws Exception {
152 List<String> subNodes = zk.getChildren(LOCK_ROOT_PATH, false);
153 // 根据元素按字典序升序排序
154 Collections.sort(subNodes);
155 System.err.println(Thread.currentThread().getName() + "创建的临时节点名称:" + selfPath.substring(LOCK_ROOT_PATH.length()+1));
156 int index = subNodes.indexOf(selfPath.substring(LOCK_ROOT_PATH.length()+1));
157 System.err.println(Thread.currentThread().getName() + "创建的临时节点的index:" + index);
158 switch (index){
159 case -1: {
160 System.err.println(Thread.currentThread().getName() + "创建的节点已不在了..." + selfPath);
161 return false;
162 }
163 case 0:{
164 System.out.println(Thread.currentThread().getName() + "子节点中,我果然是老大" + selfPath);
165 return true;
166 }
167 default:{
168 // 获取比当前节点小的前置节点,此处只关注前置节点是否还在存在,避免惊群现象产生
169 waitPath = LOCK_ROOT_PATH +"/"+ subNodes.get(index - 1);
170 System.out.println(Thread.currentThread().getName() + "获取子节点中,排在我前面的节点是:" + waitPath);
171 try {
172 zk.getData(waitPath, true, new Stat());
173 return false;
174 } catch (Exception e) {
175 if (zk.exists(waitPath, false) == null) {
176 System.out.println(Thread.currentThread().getName() + "子节点中,排在我前面的" + waitPath + "已失踪,该我了");
177 return checkMinPath();
178 } else {
179 throw e;
180 }
181 }
182 }
183
184 }
185 }
186 }
复制代码
第三步,使用memcached的cas()方法,用于分布式锁:
下篇文章我们再细说!
第四步,使用redis的watch、multi、exec命令,用于分布式锁:
下篇文章我们再细说!
第五步,总结:
综上,对于分布式锁这些非常用或者实现起来比较重的方案,大家可以根据自己在项目中的需要,酌情使用。最近在和别人讨论的过程中,以及我的第一篇关于分布式锁的文章分布式锁1 Java常用技术方案 大家的回复中,总结来看,对于用redis实现分布式锁确实存在着比较多的细节问题可以进行深入讨论,欢迎大家留言,相互学习。
忍不住嘚瑟一下,我媳妇儿此刻在我旁边看AbstractQueuedSynchronizer,厉害吧?!,一会儿出去吃饭,哈哈~
第六步,线上使用补充篇:
截止到2017.08.25(周五),使用上述文章中的”临时节点+watcher机制方案”解决一个分布式锁的问题时,最终发现在实现过程中,由于watcher机制类似于通知等待机制的特点,如果主线程在经历“获取锁操作”、“处理业务代码”、“释放锁操作”这三步的过程中,使用watcher机制阻塞的获取锁时,会导致根本无法将获取锁结果返回给主线程,而在实际的时候过程中,一般情况下主线程在“获取锁操作”时都希望可以同步获得一个返回值。
所以,上述的”临时节点+watcher机制方案”从技术方案角度足够完美,但是在实际使用过程中,个人觉得还不是特别的方便。
转载请注明来自博客园http://www.cnblogs.com/PurpleDream/p/5573040.html ,版权归本人和博客园所有,谢谢!
前言:
由于在平时的工作中,线上服务器是分布式多台部署的,经常会面临解决分布式场景下数据一致性的问题,那么就要利用分布式锁来解决这些问题。所以自己结合实际工作中的一些经验和网上看到的一些资料,做一个讲解和总结。希望这篇文章可以方便自己以后查阅,同时要是能帮助到他人那也是很好的。
===============================================================长长的分割线====================================================================
正文:
第一步,自身的业务场景:
在我日常做的项目中,目前涉及了以下这些业务场景:
场景一: 比如分配任务场景。在这个场景中,由于是公司的业务后台系统,主要是用于审核人员的审核工作,并发量并不是很高,而且任务的分配规则设计成了通过审核人员每次主动的请求拉取,然后服务端从任务池中随机的选取任务进行分配。这个场景看到这里你会觉得比较单一,但是实际的分配过程中,由于涉及到了按用户聚类的问题,所以要比我描述的复杂,但是这里为了说明问题,大家可以把问题简单化理解。那么在使用过程中,主要是为了避免同一个任务同时被两个审核人员获取到的问题。我最终使用了基于数据库资源表的分布式锁来解决的问题。
场景二: 比如支付场景。在这个场景中,我提供给用户三个用于保护用户隐私的手机号码(这些号码是从运营商处获取的,和真实手机号码看起来是一样的),让用户选择其中一个进行购买,用户购买付款后,我需要将用户选择的号码分配给用户使用,同时也要将没有选择的释放掉。在这个过程中,给用户筛选的号码要在一定时间内(用户筛选正常时间范围内)让当前用户对这个产品具有独占性,以便保证付款后是100%可以拿到;同时由于产品资源池的资源有限,还要保持资源的流动性,即不能让资源长时间被某个用户占用着。对于服务的设计目标,一期项目上线的时候至少能够支持峰值qps为300的请求,同时在设计的过程中要考虑到用户体验的问题。我最终使用了memecahed的add()方法和基于数据库资源表的分布式锁来解决的问题。
场景三: 我有一个数据服务,每天调用量在3亿,每天按86400秒计算的qps在4000左右,由于服务的白天调用量要明显高于晚上,所以白天下午的峰值qps达到6000的,一共有4台服务器,单台qps要能达到3000以上。我最终使用了redis的setnx()和expire()的分布式锁解决的问题。
场景四:场景一和场景二的升级版。在这个场景中,不涉及支付。但是由于资源分配一次过程中,需要保持涉及一致性的地方增加,而且一期的设计目标要达到峰值qps500,所以需要我们对场景进一步的优化。我最终使用了redis的setnx()、expire()和基于数据库表的分布式锁来解决的问题。
看到这里,不管你觉得我提出的业务场景qps是否足够大,都希望你能继续看下去,因为无论你身处一个什么样的公司,最开始的工作可能都需要从最简单的做起。不要提阿里和腾讯的业务场景qps如何大,因为在这样的大场景中你未必能亲自参与项目,亲自参与项目未必能是核心的设计者,是核心的设计者未必能独自设计。如果能真能满足以上三条,关闭页面可以不看啦,如果不是的话,建议还是看完,我有说的不足的地方欢迎提出建议,我说的好的地方,也希望给我点个赞或者评论一下,算是对我最大的鼓励哈。
第二步,分布式锁的解决方式:
1. 首先明确一点,有人可能会问是否可以考虑采用ReentrantLock来实现,但是实际上去实现的时候是有问题的,ReentrantLock的lock和unlock要求必须是在同一线程进行,而分布式应用中,lock和unlock是两次不相关的请求,因此肯定不是同一线程,因此导致无法使用ReentrantLock。
2. 基于数据库表做乐观锁,用于分布式锁。
3. 使用memcached的add()方法,用于分布式锁。
4. 使用memcached的cas()方法,用于分布式锁。(不常用)
5. 使用redis的setnx()、expire()方法,用于分布式锁。
6. 使用redis的setnx()、get()、getset()方法,用于分布式锁。
7. 使用redis的watch、multi、exec命令,用于分布式锁。(不常用)
8. 使用zookeeper,用于分布式锁。(不常用)
第三步,基于数据库资源表做乐观锁,用于分布式锁:
1. 首先说明乐观锁的含义:
大多数是基于数据版本(version)的记录机制实现的。何谓数据版本号?即为数据增加一个版本标识,在基于数据库表的版本解决方案中,一般是通过为数据库表添加一个 “version”字段来实现读取出数据时,将此版本号一同读出,之后更新时,对此版本号加1。
在更新过程中,会对版本号进行比较,如果是一致的,没有发生改变,则会成功执行本次操作;如果版本号不一致,则会更新失败。
2. 对乐观锁的含义有了一定的了解后,结合具体的例子,我们来推演下我们应该怎么处理:
(1). 假设我们有一张资源表,如下图所示: t_resource , 其中有6个字段id, resoource, state, add_time, update_time, version,分别表示表主键、资源、分配状态(1未分配 2已分配)、资源创建时间、资源更新时间、资源数据版本号。
(4). 假设我们现在我们对id=5780这条数据进行分配,那么非分布式场景的情况下,我们一般先查询出来state=1(未分配)的数据,然后从其中选取一条数据可以通过以下语句进行,如果可以更新成功,那么就说明已经占用了这个资源
update t_resource set state=2 where state=1 and id=5780。
(5). 如果在分布式场景中,由于数据库的update操作是原子是原子的,其实上边这条语句理论上也没有问题,但是这条语句如果在典型的“ABA”情况下,我们是无法感知的。有人可能会问什么是“ABA”问题呢?大家可以网上搜索一下,这里我说简单一点就是,如果在你第一次select和第二次update过程中,由于两次操作是非原子的,所以这过程中,如果有一个线程,先是占用了资源(state=2),然后又释放了资源(state=1),实际上最后你执行update操作的时候,是无法知道这个资源发生过变化的。也许你会说这个在你说的场景中应该也还好吧,但是在实际的使用过程中,比如银行账户存款或者扣款的过程中,这种情况是比较恐怖的。
(6). 那么如果使用乐观锁我们如何解决上边的问题呢?
a. 先执行select操作查询当前数据的数据版本号,比如当前数据版本号是26:
select id, resource, state,version from t_resource where state=1 and id=5780;
b. 执行更新操作:
update t_resoure set state=2, version=27, update_time=now() where resource=xxxxxx and state=1 and version=26
c. 如果上述update语句真正更新影响到了一行数据,那就说明占位成功。如果没有更新影响到一行数据,则说明这个资源已经被别人占位了。
3. 通过2中的讲解,相信大家已经对如何基于数据库表做乐观锁有有了一定的了解了,但是这里还是需要说明一下基于数据库表做乐观锁的一些缺点:
(1). 这种操作方式,使原本一次的update操作,必须变为2次操作: select版本号一次;update一次。增加了数据库操作的次数。
(2). 如果业务场景中的一次业务流程中,多个资源都需要用保证数据一致性,那么如果全部使用基于数据库资源表的乐观锁,就要让每个资源都有一张资源表,这个在实际使用场景中肯定是无法满足的。而且这些都基于数据库操作,在高并发的要求下,对数据库连接的开销一定是无法忍受的。
(3). 乐观锁机制往往基于系统中的数据存储逻辑,因此可能会造成脏数据被更新到数据库中。在系统设计阶段,我们应该充分考虑到这些情况出现的可能性,并进行相应调整,如将乐观锁策略在数据库存储过程中实现,对外只开放基于此存储过程的数据更新途径,而不是将数据库表直接对外公开。
4. 讲了乐观锁的实现方式和缺点,是不是会觉得不敢使用乐观锁了呢???当然不是,在文章开头我自己的业务场景中,场景1和场景2的一部分都使用了基于数据库资源表的乐观锁,已经很好的解决了线上问题。所以大家要根据的具体业务场景选择技术方案,并不是随便找一个足够复杂、足够新潮的技术方案来解决业务问题就是好方案?!比如,如果在我的场景一中,我使用zookeeper做锁,可以这么做,但是真的有必要吗???答案觉得是没有必要的!!!
第四步,使用memcached的add()方法,用于分布式锁:
对于使用memcached的add()方法做分布式锁,这个在互联网公司是一种比较常见的方式,而且基本上可以解决自己手头上的大部分应用场景。在使用这个方法之前,只要能搞明白memcached的add()和set()的区别,并且知道为什么能用add()方法做分布式锁就好。如果还不知道add()和set()方法,请直接百度吧,这个需要自己了解一下。
我在这里想说明的是另外一个问题,人们在关注分布式锁设计的好坏时,还会重点关注这样一个问题,那就是是否可以避免死锁问题???!!!
如果使用memcached的add()命令对资源占位成功了,那么是不是就完事儿了呢?当然不是!我们需要在add()的使用指定当前添加的这个key的有效时间,如果不指定有效时间,正常情况下,你可以在执行完自己的业务后,使用delete方法将这个key删除掉,也就是释放了占用的资源。但是,如果在占位成功后,memecached或者自己的业务服务器发生宕机了,那么这个资源将无法得到释放。所以通过对key设置超时时间,即便发生了宕机的情况,也不会将资源一直占用,可以避免死锁的问题。
第五步,使用memcached的cas()方法,用于分布式锁:
下篇文章我们再细说!
第六步,使用redis的setnx()、expire()方法,用于分布式锁:
对于使用redis的setnx()、expire()来实现分布式锁,这个方案相对于memcached()的add()方案,redis占优势的是,其支持的数据类型更多,而memcached只支持String一种数据类型。除此之外,无论是从性能上来说,还是操作方便性来说,其实都没有太多的差异,完全看你的选择,比如公司中用哪个比较多,你就可以用哪个。
首先说明一下setnx()命令,setnx的含义就是SET if Not Exists,其主要有两个参数 setnx(key, value)。该方法是原子的,如果key不存在,则设置当前key成功,返回1;如果当前key已经存在,则设置当前key失败,返回0。但是要注意的是setnx命令不能设置key的超时时间,只能通过expire()来对key设置。
具体的使用步骤如下:
1. setnx(lockkey, 1) 如果返回0,则说明占位失败;如果返回1,则说明占位成功
2. expire()命令对lockkey设置超时时间,为的是避免死锁问题。
3. 执行完业务代码后,可以通过delete命令删除key。
这个方案其实是可以解决日常工作中的需求的,但从技术方案的探讨上来说,可能还有一些可以完善的地方。比如,如果在第一步setnx执行成功后,在expire()命令执行成功前,发生了宕机的现象,那么就依然会出现死锁的问题,所以如果要对其进行完善的话,可以使用redis的setnx()、get()和getset()方法来实现分布式锁。
第七步,使用redis的setnx()、get()、getset()方法,用于分布式锁:
这个方案的背景主要是在setnx()和expire()的方案上针对可能存在的死锁问题,做了一版优化。
那么先说明一下这三个命令,对于setnx()和get()这两个命令,相信不用再多说什么。那么getset()命令?这个命令主要有两个参数 getset(key,newValue)。该方法是原子的,对key设置newValue这个值,并且返回key原来的旧值。假设key原来是不存在的,那么多次执行这个命令,会出现下边的效果:
1. getset(key, "value1") 返回nil 此时key的值会被设置为value1
2. getset(key, "value2") 返回value1 此时key的值会被设置为value2
3. 依次类推!
介绍完要使用的命令后,具体的使用步骤如下:
1. setnx(lockkey, 当前时间+过期超时时间) ,如果返回1,则获取锁成功;如果返回0则没有获取到锁,转向2。
2. get(lockkey)获取值oldExpireTime ,并将这个value值与当前的系统时间进行比较,如果小于当前系统时间,则认为这个锁已经超时,可以允许别的请求重新获取,转向3。
3. 计算newExpireTime=当前时间+过期超时时间,然后getset(lockkey, newExpireTime) 会返回当前lockkey的值currentExpireTime。
4. 判断currentExpireTime与oldExpireTime 是否相等,如果相等,说明当前getset设置成功,获取到了锁。如果不相等,说明这个锁又被别的请求获取走了,那么当前请求可以直接返回失败,或者继续重试。
5. 在获取到锁之后,当前线程可以开始自己的业务处理,当处理完毕后,比较自己的处理时间和对于锁设置的超时时间,如果小于锁设置的超时时间,则直接执行delete释放锁;如果大于锁设置的超时时间,则不需要再锁进行处理。
注意: 这个方案我当初在线上使用的时候是没有问题的,所以当初写这篇文章时也认为是没有问题的。但是截止到2017.05.13(周六),自己在重新回顾这篇文章时,看了文章下网友的很多评论,我发现有两个问题比较集中:
问题1: 在“get(lockkey)获取值oldExpireTime ”这个操作与“getset(lockkey, newExpireTime) ”这个操作之间,如果有N个线程在get操作获取到相同的oldExpireTime后,然后都去getset,会不会返回的newExpireTime都是一样的,都会是成功,进而都获取到锁???
我认为这套方案是不存在这个问题的。依据有两条: 第一,redis是单进程单线程模式,串行执行命令。 第二,在串行执行的前提条件下,getset之后会比较返回的currentExpireTime与oldExpireTime 是否相等。
问题2: 在“get(lockkey)获取值oldExpireTime ”这个操作与“getset(lockkey, newExpireTime) ”这个操作之间,如果有N个线程在get操作获取到相同的oldExpireTime后,然后都去getset,假设第1个线程获取锁成功,其他锁获取失败,但是获取锁失败的线程它发起的getset命令确实执行了,这样会不会造成第一个获取锁的线程设置的锁超时时间一直在延长???
我认为这套方案确实存在这个问题的可能。但我个人认为这个微笑的误差是可以忽略的,不过技术方案上存在缺陷,大家可以自行抉择哈。
第八步,使用redis的watch、multi、exec命令,用于分布式锁:
下篇文章我们再细说!
第九步,使用zookeeper,用于分布式锁:
下篇文章我们再细说!
第十步,总结:
综上,关于分布式锁的第一篇文章我就写到这儿了,在文章中主要说明了日常项目中会比较常用到四种方案,大家掌握了这四种方案,其实在日常的工作中就可以解决很多业务场景下的分布式锁的问题。从文章开头我自己的实际使用中,也可以看到,这么说完全是有一定的依据。对于另外那三种方案,我会在下一篇关于分布式锁的文章中,和大家再探讨一下。
常用的四种方案:
1. 基于数据库表做乐观锁,用于分布式锁。
2. 使用memcached的add()方法,用于分布式锁。
3. 使用redis的setnx()、expire()方法,用于分布式锁。
4. 使用redis的setnx()、get()、getset()方法,用于分布式锁。
不常用但是可以用于技术方案探讨的:
1. 使用memcached的cas()方法,用于分布式锁。
2. 使用redis的watch、multi、exec命令,用于分布式锁。
3. 使用zookeeper,用于分布式锁。
转载请注明来自博客园http://www.cnblogs.com/PurpleDream/p/5559352.html ,版权归本人和博客园所有,谢谢!
分布式锁2 Java非常用技术方案探讨之ZooKeeper
前言:
由于在平时的工作中,线上服务器是分布式多台部署的,经常会面临解决分布式场景下数据一致性的问题,那么就要利用分布式锁来解决这些问题。以自己结合实际工作中的一些经验和网上看到的一些资料,做一个讲解和总结。之前我已经写了一篇关于分布式锁的文章: 分布式锁1 Java常用技术方案 。上一篇文章中主要写的是在日常项目中,较为常见的几种实现分布式锁的方法。通过这些方法,基本上可以解决我们日常工作中大部分场景下使用分布式锁的问题。
本篇文章主要是在上一篇文章的基础上,介绍一些虽然日常工作中不常用或者比较实现起来比较重,但是可以作为技术方案学习了解一下的分布式锁方案。希望这篇文章可以方便自己以后查阅,同时要是能帮助到他人那也是很好的。
===============================================================长长的分割线====================================================================
正文:
第一步,使用zookeeper节点名称唯一性,用于分布式锁:
关于zookeeper集群的搭建,可以参考我之前写的一篇文章: ZooKeeper1 利用虚拟机搭建自己的ZooKeeper集群
zookeeper抽象出来的节点结构是一个和文件系统类似的小型的树状的目录结构,同时zookeeper机制规定:同一个目录下只能有一个唯一的文件名。例如:我们在zookeeper的根目录下,由两个客户端同时创建一个名为/myDistributeLock,只有一个客户端可以成功。
上述方案和memcached的add()方法、redis的setnx()方法实现分布式锁有着相同的思路。这样的方案实现起来如果不考虑搭建和维护zookeeper集群的成本,由于正确性和可靠性是zookeeper机制自己保证的,实现还是比较简单的。
第二步,使用zookeeper临时顺序节点,用于分布式锁:
在讨论这套方案之前,我们有必要先“吹毛求疵”般的说明一下使用zookeeper节点名称唯一性来做分布式锁这个方案的缺点。比如,当许多线程在等待一个锁时,如果锁得到释放的时候,那么所有客户端都被唤醒,但是仅仅有一个客户端得到锁。在这个过程中,大量的线程根本没有获得锁的可能性,但是也会引起大量的上下文切换,这个系统开销也是不小的,对于这样的现象有一个专业名词,称之为“惊群效应”。
我们首先说明一下zookeeper的顺序节点、临时节点和watcher机制:
所谓顺序节点,假如我们在/myDisLocks/目录下创建3个节点,zookeeper集群会按照发起创建的顺序来创建节点,节点分别为/myDisLocks/0000000001、/myDisLocks/0000000002、/myDisLocks/0000000003。
所谓临时节点,临时节点由某个客户端创建,当客户端与zookeeper集群断开连接,则该节点自动被删除。
所谓对于watcher机制,大家可以参考Apache ZooKeeper Watcher机制源码解释。当然如果你之前不知道watcher机制是个什么东东,不建议你直接去看前边我提供的文章链接,这样你极有可能忘掉我们的讨论主线,即分布式锁的实现方案,而陷入到watcher机制的源码实现中。所以你也可以先看看下面的具体方案,猜测一下watcher是用来干嘛的,我这里先总结一句话做个引子: 所谓watcher机制,你可以简单一点儿理解成任何一个连接zookeeper的客户端可以通过watcher机制关注自己感兴趣的节点的增删改查,当这个节点发生增删改查的操作时,会“广播”自己的消息,所有对此感兴趣的节点可以在收到这些消息后,根据自己的业务需要执行后续的操作。
具体的使用步骤如下:
1. 每个业务线程调用create()方法创建名为“/myDisLocks/thread”的节点,需要注意的是,这里节点的创建类型需要设置为EPHEMERAL_SEQUENTIAL,即节点类型为临时顺序节点。此时/myDisLocks节点下会出现诸如/myDisLocks/thread0000000001、/myDisLocks/thread0000000002、/myDisLocks/thread0000000003这样的子节点。
2. 每个业务线程调用getChildren(“myDisLocks”)方法来获取/myDisLocks这个节点下所有已经创建的子节点。
3. 每个业务线程获取到所有子节点的路径之后,如果发现自己在步骤1中创建的节点的尾缀编号是所有节点中序号最小的,那么就认为自己获得了锁。
4. 如果在步骤3中发现自己并非是所有子节点中序号最小的,说明自己还没有获取到锁。使用watcher机制监视比自己创建节点的序列号小的节点(比自己创建的节点小的最大节点),进入等待。比如,如果当前业务线程创建的节点是/myDisLocks/thread0000000003,那么在没有获取到锁的情况下,他只需要监视/myDisLocks/thread0000000002的情况。只有当/myDisLocks/thread0000000002获取到锁并释放之后,当前业务线程才启动获取锁,这样可以避免一个业务线程释放锁之后,其他所有线程都去竞争锁,引起不必要的上下文切换,最终造成“惊群现象”。
5. 释放锁的过程相对比较简单,就是删除自己创建的那个子节点即可。
注意: 这个方案实现的分布式锁还带着一点儿公平锁的味道!为什么呢?我们在利用每个节点的序号进行排队以此来避免进群现象时,实际上所有业务线程获得锁的顺序就是自己创建节点的顺序,也就是哪个业务线程先来,哪个就可以最快获得锁。
下面贴出我自己实现的上述方案的代码:
1. 代码中有两个Java类: MyDistributedLockByZK.java和LockWatcher.java。其中MyDistributedLockByZK.java中的main函数利用线程池启动5个线程,以此来模拟多个业务线程竞争锁的情况;而LockWatcher.java定义分布式锁和实现了watcher机制。
2. 同时,我使用的zookeeper集群是自己以前利用VMWare搭建的集群,所以zookeeper链接是192.168.224.170:2181,大家可以根据替换成自己的zookeeper链接即可。
复制代码
1 public class MyDistributedLockByZK {
2 /** 线程池 **/
3 private static ExecutorService executorService = null;
4 private static final int THREAD_NUM = 5;
5 private static int threadNo = 0;
6 private static CountDownLatch threadCompleteLatch = new CountDownLatch(THREAD_NUM);
7
8 /** ZK的相关配置常量 **/
9 private static final String CONNECTION_STRING = "192.168.224.170:2181";
10 private static final int SESSION_TIMEOUT = 10000;
11 // 此变量在LockWatcher中也有一个同名的静态变量,正式使用的时候,提取到常量类中共同维护即可。
12 private static final String LOCK_ROOT_PATH = "/myDisLocks";
13
14 public static void main(String[] args) {
15 // 定义线程池
16 executorService = Executors.newFixedThreadPool(THREAD_NUM, new ThreadFactory() {
17 @Override
18 public Thread newThread(Runnable r) {
19 String name = String.format("第[%s]个测试线程", ++threadNo);
20 Thread ret = new Thread(Thread.currentThread().getThreadGroup(), r, name, 0);
21 ret.setDaemon(false);
22 return ret;
23 }
24 });
25
26 // 启动线程
27 if (executorService != null) {
28 startProcess();
29 }
30 }
31
32 /**
33 * @author zhangyi03
34 * @date 2017-5-23 下午5:57:27
35 * @description 模拟并发执行任务
36 */
37 public static void startProcess() {
38 Runnable disposeBusinessRunnable= new Thread(new Runnable() {
39 public void run() {
40 String threadName = Thread.currentThread().getName();
41
42 LockWatcher lock = new LockWatcher(threadCompleteLatch);
43 try {
44 /** 步骤1: 当前线程创建ZK连接 **/
45 lock.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
46
47 /** 步骤2: 创建锁的根节点 **/
48 // 注意,此处创建根节点的方式其实完全可以在初始化的时候由主线程单独进行根节点的创建,没有必要在业务线程中创建。
49 // 这里这样写只是一种思路而已,不必局限于此
50 synchronized (MyDistributedLockByZK.class){
51 lock.createPersistentPath(LOCK_ROOT_PATH, "该节点由" + threadName + "创建", true);
52 }
53
54 /** 步骤3: 开启锁竞争并执行任务 **/
55 lock.getLock();
56 } catch (Exception e) {
57 e.printStackTrace();
58 }
59 }
60 });
61
62 for (int i = 0; i < THREAD_NUM; i++) {
63 executorService.execute(disposeBusinessRunnable);
64 }
65 executorService.shutdown();
66
67 try {
68 threadCompleteLatch.await();
69 System.out.println("所有线程运行结束!");
70 } catch (InterruptedException e) {
71 e.printStackTrace();
72 }
73 }
74 }
复制代码
复制代码
1 public class LockWatcher implements Watcher {
2 /** 成员变量 **/
3 private ZooKeeper zk = null;
4 // 当前业务线程竞争锁的时候创建的节点路径
5 private String selfPath = null;
6 // 当前业务线程竞争锁的时候创建节点的前置节点路径
7 private String waitPath = null;
8 // 确保连接zk成功;只有当收到Watcher的监听事件之后,才执行后续的操作,否则请求阻塞在createConnection()创建ZK连接的方法中
9 private CountDownLatch connectSuccessLatch = new CountDownLatch(1);
10 // 标识线程是否执行完任务
11 private CountDownLatch threadCompleteLatch = null;
12
13 /** ZK的相关配置常量 **/
14 private static final String LOCK_ROOT_PATH = "/myDisLocks";
15 private static final String LOCK_SUB_PATH = LOCK_ROOT_PATH + "/thread";
16
17 public LockWatcher(CountDownLatch latch) {
18 this.threadCompleteLatch = latch;
19 }
20
21 @Override
22 public void process(WatchedEvent event) {
23 if (event == null) {
24 return;
25 }
26
27 // 通知状态
28 Event.KeeperState keeperState = event.getState();
29 // 事件类型
30 Event.EventType eventType = event.getType();
31
32 // 根据通知状态分别处理
33 if (Event.KeeperState.SyncConnected == keeperState) {
34 if ( Event.EventType.None == eventType ) {
35 System.out.println(Thread.currentThread().getName() + "成功连接上ZK服务器");
36 // 此处代码的主要作用是用来辅助判断当前线程确实已经连接上ZK
37 connectSuccessLatch.countDown();
38 }else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
39 System.out.println(Thread.currentThread().getName() + "收到情报,排我前面的家伙已挂,我准备再次确认我是不是最小的节点!?");
40 try {
41 if(checkMinPath()){
42 getLockSuccess();
43 }
44 } catch (Exception e) {
45 e.printStackTrace();
46 }
47 }
48 } else if ( Event.KeeperState.Disconnected == keeperState ) {
49 System.out.println(Thread.currentThread().getName() + "与ZK服务器断开连接");
50 } else if ( Event.KeeperState.AuthFailed == keeperState ) {
51 System.out.println(Thread.currentThread().getName() + "权限检查失败");
52 } else if ( Event.KeeperState.Expired == keeperState ) {
53 System.out.println(Thread.currentThread().getName() + "会话失效");
54 }
55 }
56
57 /**
58 * @author zhangyi03
59 * @date 2017-5-23 下午6:07:03
60 * @description 创建ZK连接
61 * @param connectString ZK服务器地址列表
62 * @param sessionTimeout Session超时时间
63 * @throws IOException
64 * @throws InterruptedException
65 */
66 public void createConnection(String connectString, int sessionTimeout) throws IOException, InterruptedException {
67 zk = new ZooKeeper(connectString, sessionTimeout, this);
68 // connectSuccessLatch.await(1, TimeUnit.SECONDS) 正式实现的时候可以考虑此处是否采用超时阻塞
69 connectSuccessLatch.await();
70 }
71
72 /**
73 * @author zhangyi03
74 * @date 2017-5-23 下午6:15:48
75 * @description 创建ZK节点
76 * @param path 节点path
77 * @param data 初始数据内容
78 * @param needWatch
79 * @return
80 * @throws KeeperException
81 * @throws InterruptedException
82 */
83 public boolean createPersistentPath(String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
84 if(zk.exists(path, needWatch) == null){
85 String result = zk.create( path,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
86 System.out.println(Thread.currentThread().getName() + "创建节点成功, path: " + result + ", content: " + data);
87 }
88 return true;
89 }
90
91 /**
92 * @author zhangyi03
93 * @date 2017-5-23 下午6:24:46
94 * @description 获取分布式锁
95 * @throws KeeperException
96 * @throws InterruptedException
97 */
98 public void getLock() throws Exception {
99 selfPath = zk.create(LOCK_SUB_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
100 System.out.println(Thread.currentThread().getName() + "创建锁路径:" + selfPath);
101 if(checkMinPath()){
102 getLockSuccess();
103 }
104 }
105
106 /**
107 * @author zhangyi03
108 * @date 2017-5-23 下午7:02:41
109 * @description 获取锁成功
110 * @throws KeeperException
111 * @throws InterruptedException
112 */
113 private void getLockSuccess() throws KeeperException, InterruptedException {
114 if(zk.exists(selfPath, false) == null){
115 System.err.println(Thread.currentThread().getName() + "本节点已不在了...");
116 return;
117 }
118 System.out.println(Thread.currentThread().getName() + "获取锁成功,开始处理业务数据!");
119 Thread.sleep(2000);
120 System.out.println(Thread.currentThread().getName() + "处理业务数据完成,删除本节点:" + selfPath);
121 zk.delete(selfPath, -1);
122 releaseConnection();
123 threadCompleteLatch.countDown();
124 }
125
126 /**
127 * @author zhangyi03
128 * @date 2017-5-23 下午7:06:46
129 * @description 关闭ZK连接
130 */
131 private void releaseConnection() {
132 if (zk != null) {
133 try {
134 zk.close();
135 } catch (InterruptedException e) {
136 e.printStackTrace();
137 }
138 }
139 System.out.println(Thread.currentThread().getName() + "释放ZK连接");
140 }
141
142 /**
143 * @author zhangyi03
144 * @date 2017-5-23 下午6:57:14
145 * @description 检查自己是不是最小的节点
146 * @param selfPath
147 * @return
148 * @throws KeeperException
149 * @throws InterruptedException
150 */
151 private boolean checkMinPath() throws Exception {
152 List<String> subNodes = zk.getChildren(LOCK_ROOT_PATH, false);
153 // 根据元素按字典序升序排序
154 Collections.sort(subNodes);
155 System.err.println(Thread.currentThread().getName() + "创建的临时节点名称:" + selfPath.substring(LOCK_ROOT_PATH.length()+1));
156 int index = subNodes.indexOf(selfPath.substring(LOCK_ROOT_PATH.length()+1));
157 System.err.println(Thread.currentThread().getName() + "创建的临时节点的index:" + index);
158 switch (index){
159 case -1: {
160 System.err.println(Thread.currentThread().getName() + "创建的节点已不在了..." + selfPath);
161 return false;
162 }
163 case 0:{
164 System.out.println(Thread.currentThread().getName() + "子节点中,我果然是老大" + selfPath);
165 return true;
166 }
167 default:{
168 // 获取比当前节点小的前置节点,此处只关注前置节点是否还在存在,避免惊群现象产生
169 waitPath = LOCK_ROOT_PATH +"/"+ subNodes.get(index - 1);
170 System.out.println(Thread.currentThread().getName() + "获取子节点中,排在我前面的节点是:" + waitPath);
171 try {
172 zk.getData(waitPath, true, new Stat());
173 return false;
174 } catch (Exception e) {
175 if (zk.exists(waitPath, false) == null) {
176 System.out.println(Thread.currentThread().getName() + "子节点中,排在我前面的" + waitPath + "已失踪,该我了");
177 return checkMinPath();
178 } else {
179 throw e;
180 }
181 }
182 }
183
184 }
185 }
186 }
复制代码
第三步,使用memcached的cas()方法,用于分布式锁:
下篇文章我们再细说!
第四步,使用redis的watch、multi、exec命令,用于分布式锁:
下篇文章我们再细说!
第五步,总结:
综上,对于分布式锁这些非常用或者实现起来比较重的方案,大家可以根据自己在项目中的需要,酌情使用。最近在和别人讨论的过程中,以及我的第一篇关于分布式锁的文章分布式锁1 Java常用技术方案 大家的回复中,总结来看,对于用redis实现分布式锁确实存在着比较多的细节问题可以进行深入讨论,欢迎大家留言,相互学习。
忍不住嘚瑟一下,我媳妇儿此刻在我旁边看AbstractQueuedSynchronizer,厉害吧?!,一会儿出去吃饭,哈哈~
第六步,线上使用补充篇:
截止到2017.08.25(周五),使用上述文章中的”临时节点+watcher机制方案”解决一个分布式锁的问题时,最终发现在实现过程中,由于watcher机制类似于通知等待机制的特点,如果主线程在经历“获取锁操作”、“处理业务代码”、“释放锁操作”这三步的过程中,使用watcher机制阻塞的获取锁时,会导致根本无法将获取锁结果返回给主线程,而在实际的时候过程中,一般情况下主线程在“获取锁操作”时都希望可以同步获得一个返回值。
所以,上述的”临时节点+watcher机制方案”从技术方案角度足够完美,但是在实际使用过程中,个人觉得还不是特别的方便。
转载请注明来自博客园http://www.cnblogs.com/PurpleDream/p/5573040.html ,版权归本人和博客园所有,谢谢!
发表评论
-
Flink入门到实践
2022-02-09 09:36 3721 导言 通过本文可以快 ... -
JavaAgent 应用(spring-loaded 热部署)
2021-11-16 16:26 470上一篇文章简单介绍了 javaagent ,想了解的可以移步 ... -
细分十一步,助你构建完整的数据运营体系
2020-12-15 09:26 198https://www.niaogebiji.com/arti ... -
Nginx的配置
2018-10-25 15:49 283Nginx的配置文件nginx.conf ... -
idea注册
2018-09-10 09:47 593开始 G91XMO9AVI-eyJsaWNlbnNlSWQiO ... -
java判断字符串是否为数字或中文或字母
2018-08-31 16:55 9492*各种字符的unicode编码 ... -
JAVA多线程实现的四种方式
2018-08-31 14:26 461Java多线程实现方式主要有四种:继承Thread类、实现Ru ... -
spring 注解
2017-10-23 09:59 360声明Bean的注解: @Component ... -
java内存管理与垃圾回收
2017-07-25 15:01 3031、Java虚拟机运行时的 ... -
jstat的用法
2017-07-25 10:15 546jstat的用法 用以判断JVM是否存在内存问题呢?如何判 ... -
JVM 调优参数详解
2017-07-24 14:05 338GC有两种类型:Scavenge GC 和Full GC 1、 ... -
JVM参数调优技巧
2017-07-24 14:02 409JVM参数调优实例解析 关于JVM参数调优,对于很多程序员来 ... -
Elasticsearch使用基础教程
2017-06-25 15:28 320基础概念 Elastics ... -
Quartz集成springMVC (持久化任务、集群和分布式)
2017-06-22 11:15 2204Quartz是一个开放源码项目,专注于任务调度器,提供了极为 ... -
JAVA 实现XML与JSON 相互转换
2017-06-22 09:22 18431.把XML转为JSON格式 ... -
hive语法详解
2016-09-29 16:35 439Hive 是基于Hadoop 构建的一套数据仓库分析系统,它提 ... -
使用elasticsearch遇到的一些问题以及解决方法
2016-09-21 16:14 4881.由gc引起节点脱离集群 因为gc时会使jvm停 ... -
分布式系统之消息中间件rabbitmq
2016-09-21 09:49 436既然要做分布式系统,就不得不说分布式消息通信系统。分布式系统的 ... -
强大的分布式消息中间件——kafka
2016-09-21 09:45 1278在我们大量使用分布式数据库、分布式计算集群的时候,是否会遇到这 ... -
RabbitMq、ActiveMq、ZeroMq、kafka之间的比较
2016-09-21 09:42 698MQ框架非常之多,比较 ...
相关推荐
Java基于Redis实现分布式锁代码实例 分布式锁的必要性 在多线程环境中,资源竞争是一个常见的问题。例如,在一个简单的用户操作中,一个线程修改用户状态,首先在内存中读取用户状态,然后在内存中进行修改,然后...
在IT行业中,尤其是在大型分布式系统的设计与开发中,分布式锁是一种关键的同步机制。本篇文章将深入探讨如何在C#.NET环境下利用Redis实现分布式锁,以及相关的核心知识点。 首先,让我们理解什么是分布式锁。...
在分布式系统中,为了保证数据的一致性和安全性,分布式锁是一种常见的解决方案。Redis作为一个高性能的键值存储系统,常被用作实现分布式锁的工具。本文将深入探讨如何使用Redis实现分布式锁,以及如何利用自旋式...
本文将深入探讨C#中如何使用Lock和Redis分布式锁来解决并发问题,以秒杀系统为例进行阐述。 首先,让我们理解什么是并发控制。并发控制是指在多线程环境下确保数据的一致性和完整性,防止多个线程同时访问同一资源...
分布式锁是一种在分布式系统中实现同步访问资源的关键技术。它允许多个节点在同一时间对共享资源进行操作,但确保任何时刻只有一个节点能持有锁并执行相应的操作。本视频资料深入浅出地讲解了分布式锁的原理、实现...
分布式锁是一种在分布式系统中实现资源同步的关键技术,它确保在多节点环境下,同一时间只有一个节点可以访问或修改特定的共享资源,以防止数据不一致性和并发问题。在Oracle数据库中,分布式事务处理可能会导致...
本文将深入探讨如何使用C++结合Redis实现分布式锁,并详细讲解Redis API在C++中的应用,以及如何处理与Boost库的集成。 首先,Redis是一个高性能的键值存储数据库,广泛用于缓存、消息队列、分布式锁等场景。分布式...
Redis 分布式锁是分布式系统中解决并发控制和数据一致性问题的一种常见机制。在大型分布式应用中,单机锁无法满足需求,因为它们局限于单个服务器。Redis 的高可用性和低延迟特性使其成为实现分布式锁的理想选择。...
现在很多项目单机版已经不满足了,分布式变得越受欢迎,同时也带来很多问题,分布式锁也变得没那么容易实现,分享一个redis分布式锁工具类,里面的加锁采用lua脚本(脚本比较简单,采用java代码实现,无须外部调用...
分布式锁是一种在分布式系统中实现锁机制的技术,用于在多节点之间协调访问共享资源,确保在高并发环境下数据的一致性和完整性。本压缩包“zk:redis分布式锁.zip”提供了基于Zookeeper(zk)和Redis两种分布式锁实现...
在现代的高并发系统中,分布式锁是一种非常重要的机制,用于协调多个节点间的资源访问,以确保数据的一致性和完整性。SpringBoot是一个流行的Java微服务框架,它简化了与各种技术栈的集成,包括Redis这样的高性能...
分布式锁是解决分布式系统中多个进程间共享资源互斥访问的一种方法,它保证了即使在分布式环境下,也能维持数据的一致性。随着互联网公司业务的不断扩展和技术的进步,分布式系统的数据量和业务复杂性大幅增加,...
分布式锁是分布式系统中用于同步资源访问的一种机制,它能够保证在分布式部署的应用系统中,同一时刻只允许一个客户端对共享资源进行操作。随着互联网业务的发展,分布式系统越来越多地成为企业架构的选择,因此...
在分布式系统中,锁是保证数据一致性的重要工具,而Redisson的分布式锁则为Java开发者提供了一种高效且可靠的解决方案。 首先,让我们深入理解分布式锁的概念。分布式锁是在分布式系统中,为了确保多个节点对共享...
分布式锁是一种在分布式系统中实现同步的技术,它允许多个节点在同一时刻访问共享资源。在大型分布式环境中,由于网络延迟和并发操作,简单的本地锁可能无法有效解决数据一致性问题。这时,Zookeeper,一个高可用的...
在分布式系统中,为了保证数据的一致性和安全性,分布式锁是一种常见的解决方案。本文将深入探讨如何使用Redisson和Curator框架来实现Java环境中的分布式锁。 首先,让我们来看一下Redisson实现的分布式锁。Redis是...
分布式锁是一种在分布式系统中实现同步访问资源的关键技术,它允许多台服务器共享同一资源而不会引发数据不一致。在大型互联网系统中,由于服务的水平扩展和数据的分布式存储,传统的单机锁已无法满足需求,分布式锁...
分布式锁是一种在分布式系统中实现同步访问资源的机制,它允许多个节点在同一时间对共享资源进行操作,而不会导致数据不一致或并发问题。在Java开发中,分布式锁的应用广泛,尤其是在微服务架构中,当服务间的通信...
redis实现分布式锁(java/jedis),其中包含工具方法以及使用demo 本资源是利用java的jedis实现 redis实现分布式锁(java/jedis),其中包含工具方法以及使用demo 本资源是利用java的jedis实现
面试中,分布式锁、分布式事务和分布式缓存等概念是考察开发者技术能力的重要方面。下面将对这些主题进行详细阐述。 分布式锁是解决分布式环境下的资源争抢问题的关键工具。在单机系统中,锁的实现相对简单,但在...