网畅学校网站管理系统,做网站大概花多少钱,北京丰台区,网站做优化的必要性作者#xff1a;来自 vivo 互联网服务器团队- Xu Yaoming 介绍分布式锁的实现原理。
一、分布式锁概述
分布式锁#xff0c;顾名思义#xff0c;就是在分布式环境下使用的锁。众所周知#xff0c;在并发编程中#xff0c;我们经常需要借助并发控制工具#xff0c;如 mu… 作者来自 vivo 互联网服务器团队- Xu Yaoming 介绍分布式锁的实现原理。
一、分布式锁概述
分布式锁顾名思义就是在分布式环境下使用的锁。众所周知在并发编程中我们经常需要借助并发控制工具如 mutex、synchronized 等来保障线程安全。但是这种线程安全仅作用在同一内存环境中。在实际业务中为了保障服务的可靠性我们通常会采用多节点进行部署。在这种分布式情况下各实例间的内存不共享线程安全并不能保证并发安全如下例同一实例中线程A与线程B之间的并发安全并不能保证实例1与实例2之间的并发安全 因此当遇到分布式系统的并发安全问题时我们就可能会需要引入分布式锁来解决。
用于实现分布式锁的组件通常都会具备以下的一些特性 互斥性提供分布式环境下的互斥原语来加锁/释放锁当然是分布式锁最基本的特性。 自动释放为了应对分布式系统中各实例因通信故障导致锁不能释放的问题自动释放的特性通常也是很有必要的。 分区容错性应用在分布式系统的组件具备分区容错性也是一项重要的特性否则就会成为整个系统的瓶颈。
目前开源社区中常见的分布式锁解决方案大多是基于具备集群部署能力的 key-value 存储中间件来实现最为常用的方案基本上是基于 Redis、zookeeper 来实现笔者将从上述分布式锁的特性出发介绍一下这两类的分布式锁解决方案的优缺点。
二、分布式锁的实现原理
2.1 Redis 实现分布式锁
Redis 由于其高性能、使用及部署便利性在很多场景下是实现分布式锁的首选。首先我们看下 Redis 是如何实现互斥性的。在单机部署的模式下Redis 由于其单线程处理命令的线程模型天然的具备互斥能力而在哨兵/集群模式下写命令也是单独发送到某个单独节点上进行处理可以保证互斥性其核心的命令是 set if not exist
SET lockKey lockValue NX
成功设置 lockValue 的实例就相当于抢锁成功。但如果持有锁的实例宕机因为 Redis 服务端并没有感知客户端状态的能力因此会出现锁无法释放的问题 这种情况下就需要给 key 设置一个过期时间 expireTime
SET lockKey lockValue EX expireTime NX
如果持有锁的实例宕机无法释放锁则锁会自动过期这样可以就避免锁无法释放的问题。在一些简单的场景下通过该方式实现的分布式锁已经可以满足需求。但这种方式存在一个明显问题如果业务的实际处理时间比锁过期时间长锁就会被误释放导致其他实例也可以加锁 这种情况下就需要通过其他机制来保证锁在业务处理结束后再释放一个常用的方式就是通过后台线程的方式来实现锁的自动续期。 Redssion 是开源社区中比较受欢迎的一个 Java 语言实现的 Redis 客户端其对 Java 中 Lock 接口定义进行扩展实现了 Redis 分布式锁并通过 watchDog 机制本质上即是后台线程运作来对锁进行自动续期。以下是一个简单的 Reddison 分布式锁的使用例子
RLock rLock RedissonClient.getLock(test-lock);
try {if (rLock.tryLock()) {// do something}
} finally {rLock.unlock();
}
Redssion 的默认实现 RedissonLock 为可重入互斥非公平锁其 tryLock 方法会基于三个可选参数执行 waitTime获取锁的最长等待时长默认为-1waitTime 参数决定在获取锁的过程中是否需要进行等待如果 waitTime0则在获取锁的过程中线程会等待一定时间并持续尝试获取锁否则获取锁失败会直接返回。 leaseTime锁持有时长默认为-1。当 leaseTime0 时会开启 watchDog 机制进行自动续期而 leaseTime0 时则不会进行自动续期到达 leaseTime 锁即过期释放 unit时间单位标识 waitTime 及 leaseTime 的时间单位
我们不妨通过参数最全的 RedissonLock#tryLock(long waitTime, long leaseTime, TimeUnit unit) 方法源码来一探其完整的加锁过程
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {...// tryAcquire方法返回锁的剩余有效时长ttl如果未上锁则为nullLong ttl tryAcquire(waitTime, leaseTime, unit, threadId);if (ttl null) {// 获取锁成功return true;}// 计算剩余等待时长剩余等待时长小于0则不再尝试获取锁获取锁失败后续有多处同样的判断逻辑将精简省略time - System.currentTimeMillis() - current;if (time 0) {acquireFailed(waitTime, unit, threadId);return false;}// 等待时长大于0则会对锁释放的事件进行订阅持有锁的客户端在锁释放时会发布锁释放事件通知其他客户端抢锁由此可得知该默认实现为非公平锁。// Redisson对Redis发布订阅机制的实现底层大量使用了CompletableFuture、CompletionStage等接口来编写异步回调代码感兴趣的读者可以详细了解此处不作展开CompletableFutureRedissonLockEntry subscribeFuture subscribe(threadId);try {subscribeFuture.get(time, TimeUnit.MILLISECONDS);} catch (TimeoutException e) {...} catch (ExecutionException e) {...}try {...// 循环尝试获取锁while (true) {long currentTime System.currentTimeMillis();ttl tryAcquire(waitTime, leaseTime, unit, threadId);// lock acquiredif (ttl null) {return true;}...// 此处通过信号量来将线程阻塞一定时间避免无效的申请锁浪费资源在阻塞期间如果收到了锁释放的事件则会通过信号量提前唤起阻塞线程重新尝试获取锁currentTime System.currentTimeMillis();if (ttl 0 ttl time) {// 若ttl锁过期时长小于time剩余等待时长则将线程阻塞ttlcommandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {// 若等待时长小于ttl则将线程阻塞timecommandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);}...}} finally {// 取消订阅unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);}
}
上述代码逻辑主要集中在处理 waitTime 参数在并发竞争不激烈、可以容忍一定的等待时间的情况下合理设置 waitTime 参数可以提高业务并发运行成功率避免抢锁失败直接返回错误但在并发竞争激烈、对性能有较高要求时建议不设置 waitTime或者直接使用没有 waitTime 参数的 lock() 方法通过快速失败来提高系统吞吐量。
一个比较值得注意的点是如果设置了 waitTime 参数则 Redisson 通过将 RedissonLockEntry 中信号量Semaphore的许可证数初始化为0来达到一定程度的限流保证锁释放后只有一个等待中的线程会被唤醒去请求 Redis 服务端把唤醒等待线程的工作分摊到各个客户端实例上可以很大程度上缓解非公平锁给 Redis 服务端带来的惊群效应压力。
public class RedissonLockEntry implements PubSubEntryRedissonLockEntry {...private final Semaphore latch;public RedissonLockEntry(CompletableFutureRedissonLockEntry promise) {super();// RedissonLockEntry 中的Semaphore的许可证数初始化为0this.latch new Semaphore(0);this.promise promise;}...
}
获取锁的核心逻辑会通过 RedissonLock#tryAcquire 方法调用到 RedissonLock#tryAcquireAsync 方法。
private RFutureLong tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {RFutureLong ttlRemainingFuture;if (leaseTime 0) {// 若leaseTime大于零会设置锁的租期为leaseTimettlRemainingFuture tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);} else {// 若leaseTime小于或等于零会设置锁的租期为internalLockLeaseTime这是一个通过lockWatchdogTimeout配置的值默认为30sttlRemainingFuture tryLockInnerAsync(waitTime, internalLockLeaseTime,TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);}// 此处的handleNoSync方法是为了解决Redis发生故障转移集群拓扑改变后只有持有锁的客户端能再次获得锁的bug为3.20.1版本修复详见Redisson issue#4822CompletionStageLong s handleNoSync(threadId, ttlRemainingFuture);ttlRemainingFuture new CompletableFutureWrapper(s);// 根据加锁情况来进行后续处理CompletionStageLong f ttlRemainingFuture.thenApply(ttlRemaining - {// lock acquired// 若ttl为空说明加锁不成功if (ttlRemaining null) {if (leaseTime 0) {// 若leaseTime0则将internalLockLeaseTime变量设置为leaseTime以便后续解锁使用internalLockLeaseTime unit.toMillis(leaseTime);} else {// 若leaseTime0则开启看门狗机制通过定时任务进行锁续期scheduleExpirationRenewal(threadId);}}return ttlRemaining;});return new CompletableFutureWrapper(f);
}// 加锁的lua脚本
T RFutureT tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommandT command) {return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,if ((Redis.call(exists, KEYS[1]) 0) or (Redis.call(hexists, KEYS[1], ARGV[2]) 1)) then Redis.call(hincrby, KEYS[1], ARGV[2], 1); Redis.call(pexpire, KEYS[1], ARGV[1]); return nil; end; return Redis.call(pttl, KEYS[1]);,Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}
可以看到若 leaseTime 大于0则不会开启看门狗机制锁在过期后即失效在使用时请务必留意。上述代码中执行的 scheduleExpirationRenewal 方法即为看门狗机制的实现逻辑
protected void scheduleExpirationRenewal(long threadId) {// 每个锁都会对应一个ExpirationEntry类第一次加锁时不存在oldEntryExpirationEntry new ExpirationEntry();ExpirationEntry oldEntry EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);if (oldEntry ! null) {// 非首次加锁重入计数不作其他操作oldEntry.addThreadId(threadId);} else {// 首次加锁调用renewExpiration()方法进行自动续期entry.addThreadId(threadId);try {renewExpiration();} finally {// 若当前线程被中断则取消对锁的自动续期。if (Thread.currentThread().isInterrupted()) {cancelExpirationRenewal(threadId);}}}
}private void renewExpiration() {...// 此处使用的是netty的时间轮来执行定时续期此处不对时间轮做展开感兴趣的读者可详细了解Timeout task getServiceManager().newTimeout(new TimerTask() {Overridepublic void run(Timeout timeout) throws Exception {...CompletionStageBoolean future renewExpirationAsync(threadId);future.whenComplete((res, e) - {if (e ! null) {log.error(Cant update lock {} expiration, getRawName(), e);EXPIRATION_RENEWAL_MAP.remove(getEntryName());return;}if (res) {// 若续期成功则递归调用等待任务的下一次执行renewExpiration();} else {// 若续期结果为false说明锁已经过期了或锁易主了则清理当前线程关联的信息等待线程结束cancelExpirationRenewal(null);}});}// 时间轮的执行周期为internalLockLeaseTime / 3即默认情况下internalLockLeaseTime为30s时每10s触发一次自动续期}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}protected CompletionStageBoolean renewExpirationAsync(long threadId) {// 执行重置过期时间的lua脚本return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,if (Redis.call(hexists, KEYS[1], ARGV[2]) 1) then Redis.call(pexpire, KEYS[1], ARGV[1]); return 1; end; return 0;,Collections.singletonList(getRawName()),internalLockLeaseTime, getLockName(threadId));
}
上面一段代码即是看门狗调度的核心代码本质上即是通过定时调度线程执行 lua 脚本来进行锁续期。值得留意的是 scheduleExpirationRenewal
方法中的 ExpirationEntry该对象与锁一一关联会存储尝试获取该锁的线程无论是否获取成功以及重入锁的次数在锁失效/锁释放时会根据该对象中存储的线程逐一进行资源释放操作以保证资源的正确释放。
最后对上述 Redisson 可重入非公平锁源码进行一下总结 Redisson 加锁时根据 waitTime 参数是否大于0来决定加锁失败时采用等待并再次尝试/快速失败的策略 Redisson 加锁时根据 leaseTime 参数是否小于等于0来决定是否开启看门狗机制进行定时续期 Redisson 底层使用了 netty 实现的时间轮来进行定时续期任务的调度执行周期为 internalLockLeaseTime / 3默认为10s。
2.2 zookeeper 实现分布式锁
zookeeper后文均简称 zk 基于 zab 协议实现的分布式协调服务天生具备实现分布式锁的基础条件。我们可以从zk的一些基本机制入手了解其是如何实现分布式锁的。 zab为了保证分布式一致性zk 实现了 zabZk Atomic Broadcastzk 原子广播协议在 zab 协议下zk集群分为 Leader 节点及 Follower 节点其中负责处理写请求的 Leader 节点在集群中是唯一的多个 Follower 则负责同步 Leader 节点的数据处理客户端的读请求。同时zk 处理写请求时底层数据存储使用的是 ConcurrentHashMap以保证并发安全
public class NodeHashMapImpl implements NodeHashMap {private final ConcurrentHashMapString, DataNode nodes;private final boolean digestEnabled;private final DigestCalculator digestCalculator;private final AdHash hash;...}
临时顺序节点zk 的数据呈树状结构树上的每一个节点为一个基本数据单元称为 Znode。zk 可以创建一类临时顺序EPHEMERAL_SEQUENTIAL节点在满足一定条件时会可以自动释放同时同一层级的节点名称会按节点的创建顺序进行命名第一个节点为xxx-0000000000第二个节点则为xxx-0000000001以此类推 sessionzk 的服务端与客户端使用 session 机制进行通信简单来说即是通过长连接来进行交互zk 服务端会通过心跳来监控客户端是否处于活动状态。若客户端长期无心跳或断开连接则 zk 服务端会定期关闭这些 session主动断开与客户端的通信。
了解了上述 zk 特点我们不难发现 zk 也是具备互斥性、自动释放的特性的。同时zk 由于 session 机制的存在服务端可以感知到客户端的状态因此不需要有由客户端来进行节点续期zk 服务端可以主动地清理失联客户端创建的节点避免锁无法释放的问题。zk 实现分布式锁的主要步骤如下 client1 申请加锁创建 /lock/xxx-lock-0000000000节点临时顺序节点并监听其父节点 /lock client1 查询 /lock 节点下的节点列表并判断自己创建的 /xxx-lock-0000000000 是否为 /lock 节点下的第一个节点当前没有其他客户端加锁所以 client1 获取锁成功 若 client2 此时来加锁则会创建 /lock/xxx-lock-0000000001 节点此时 client2 查询 /lock 节点下的节点列表此时 /xxx-lock-0000000001 并非 /lock 下的第一个节点因此加锁不成功此时 client2 则会监听其上一个节点 /xxx-lock-0000000000 client1 释放锁client1 删除 /xxx-lock-0000000000 节点zk 服务端通过长连接 session 通知监听了 /xxx-lock-0000000000 节点的 client2 来获取锁 收到释放事件的 client2 查询 /lock 节点下的节点列表此时自己创建的 /xxx-lock-0000000001 为最小节点因此获取锁成功。 上述是 zk 公平锁的一种常见实现方式。值得注意的是 zk 客户端通常并不会实现非公平锁。事实上zk 上锁的粒度不局限于上述步骤中的客户端zk 客户端每次获取锁请求即每一个尝试获取锁的线程都会向 zk 服务端请求创建一个临时顺序节点。
以上述步骤为例如果需要实现非公平锁则会导致其余的所有节点都需要监听第一个节点 /xxx-lock-0000000000 的释放事件相当于所有等待锁释放的线程都会监听同一个节点这种机制无法像 Redisson 一样把唤醒锁的压力分摊到客户端上或者说实现起来比较困难会产生比较严重的惊群效应因此使用 zk 实现的分布式锁一般情况下都是公平锁。
Curator 是一个比较常用的 zk 客户端我们可以通过 Curator 的加锁过程来了解 zk 分布式锁的设计原理。Curator 中比较常用的是可重入互斥公平锁 InterProcessMutex
InterProcessMutex mutex new InterProcessMutex(zkClient, /lock);
try {// acquire方法的两个参数等待时长及时间单位if (mutex.acquire(3, TimeUnit.SECONDS)) {log.info(加锁成功);} else {log.info(加锁失败);}
} finally {mutex.release();
}
InterProcessMutex 同样提供了等待时长参数用于设置没有立即获取到锁时是快速失败还是阻塞等待下一步方法会调用到 InterProcessMutex#internalLock 方法中
private boolean internalLock(long time, TimeUnit unit) throws Exception
{// 注释的意思一个LockData对象只会被一个持有锁的线程进行修改因此不需要对LockData进行并发控制。如此说明的原因是zk的互斥特性保证了下方attemptLock方法的互斥由此保证了LockData不会被并发修改/*Note on concurrency: a given lockData instancecan be only acted on by a single thread so locking isnt necessary*/Thread currentThread Thread.currentThread();// LockData用于记录当前持有锁的线程数据LockData lockData threadData.get(currentThread);if ( lockData ! null ){// 线程不为空则进行重入重入次数1// re-enteringlockData.lockCount.incrementAndGet();return true;}// 向zk服务获取分布式锁getLockNodeBytesString lockPath internals.attemptLock(time, unit, getLockNodeBytes());if ( lockPath ! null ){// 若lockPath不为空则获取锁成功记录当前持有锁的线程LockData newLockData new LockData(currentThread, lockPath);threadData.put(currentThread, newLockData);return true;}return false;
}
InterProcessMutex#internalLock会调用到 LockInternals#attemptLock 方法
String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
{...while ( !isDone ){isDone true;try{// 创建锁节点ourPath driver.createsTheLock(client, path, localLockNodeBytes);// 判断是否成功获取锁hasTheLock internalLockLoop(startMillis, millisToWait, ourPath);}catch ( KeeperException.NoNodeException e ){// 捕获由于网络中断、session过期等原因导致的无法获得节点异常此处根据配置的zk客户端重试策略决定是否重试默认重试策略为Exponential Backoff...retry or not...}}if ( hasTheLock ){return ourPath;}return null;
}public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception
{String ourPath;if ( lockNodeBytes ! null ){ // 在其他类型的锁实现中lockNodeBytes可能不为空则根据lockNodeBytes来获取节点路径此处暂不作展开ourPath client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, lockNodeBytes);}else{// 在可重入互斥锁中客户端向zk服务端请求创建一个 EPHEMERAL_SEQUENTIAL 临时顺序节点ourPath client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);}return ourPath;
}
上述代码中创建锁节点并不会产生互斥而是会直接向 zk 服务端请求创建临时顺序节点。此时客户端还未真正的获得锁判断加锁成功的核心逻辑在 LockInternals#internalLockLoop 方法中
private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
{boolean haveTheLock false;boolean doDelete false;try{if ( revocable.get() ! null ){ // curator锁撤销机制通过实现Curator中的Revocable接口的makeRevocable方法可以将锁设置为可撤销锁其他线程可以在符合条件时将锁撤销此处暂不涉及client.getData().usingWatcher(revocableWatcher).forPath(ourPath);}// 客户端实例就绪则尝试循环获取锁while ( (client.getState() CuratorFrameworkState.STARTED) !haveTheLock ) {// 获取当前父节点下的排好序的子节点ListString children getSortedChildren();// 得到当前节点名String sequenceNodeName ourPath.substring(basePath.length() 1); // 1 to include the slash// 根据 children 列表与当前节点名计算当前节点是否为第一个节点若不是第一个节点则在 PredicateResults中返回需要监听的前一个节点节点若为最小节点则获取锁成功PredicateResults predicateResults driver.getsTheLock(client, children, sequenceNodeName, maxLeases);if ( predicateResults.getsTheLock() ){// 获取锁成功haveTheLock true;}else{// 拼接前一个节点的节点路径String previousSequencePath basePath / predicateResults.getPathToWatch();synchronized(this){try{// 将前一个节点的监听器放到当前客户端中当前一个节点被释放时就会唤醒当前客户端client.getData().usingWatcher(watcher).forPath(previousSequencePath);if ( millisToWait ! null ){millisToWait - (System.currentTimeMillis() - startMillis);startMillis System.currentTimeMillis();// 计算剩余等待时长若等待时长小于0则不再尝试获取锁并标记当前线程创建的节点需要删除if ( millisToWait 0 ){doDelete true; // timed out - delete our nodebreak;}// 若等待时长大于0则阻塞线程等待锁释放wait(millisToWait);}else{// 在其他的一些加锁场景中默认会持久等待到锁释放位置当前可重入互斥锁暂不涉及wait();}}catch ( KeeperException.NoNodeException e ){// it has been deleted (i.e. lock released). Try to acquire again}}}}}catch ( Exception e ){ThreadUtils.checkInterrupted(e);doDelete true;throw e;}finally{if ( doDelete ){// 删除当前节点deleteOurPath(ourPath);}}return haveTheLock;
}private synchronized void notifyFromWatcher()
{// 当zk客户端收到锁释放事件时会遍历当前客户端注册过的所有的监听器并找到合适的监听器进行回调最终通过notifyAll唤醒监听被释放节点的线程notifyAll();
}
上述 curator 加锁的核心代码虽然比较长但整体逻辑与我们前面分析过的加锁逻辑是一致的主要做了三件事 获取当前父节点的有序子节点序列 判断当前节点是否为第一个节点 若为第一个节点则获取锁成功否则为当前 zk 客户端增加一个前一节点的监听器如果此时还在等待时长内则使用wait方法挂起线程否则删除当前节点。
三、总结——如何选择合适的分布式并发安全解决方案 绕不过的 CAP 理论
Redis 与 zk 由于客户端与服务端的交互机制上存在比较大的差异相应的分布式锁实现原理也有所不同。两者都是优秀的支持分布式部署的系统自然具备分区容错性但分布式系统总绕不过去一个经典的问题——CAP理论在满足了分区容错性的前提下分布式系统只能满足可用性、数据一致性两者其一。 对比之下Redis 在可用性上更胜一筹属于 AP 系统zk 具备更强的数据一致性属于 CP 系统而基于 AP、CP 的特性去实现的分布式锁自然也会存在不同程度的问题。 Redis 分布式锁的一致性问题
Redis 的集群模式并没有严格地实现分布式共识算法因此 Redis 是不具备一致性的。为了保证高可用性Redis 集群的主从节点使用的是异步复制从节点并不保证与主节点数据一致只能尽量的追赶主节点的最新数据因此当主节点发生故障进行主从切换时实际上有可能会发生数据丢失问题 zk 性能及可用性问题
zk 实现了 zab 算法在数据一致性上给出了比较可靠的方案但是由于 zab 协议的两阶段提交要求所有节点的写请求处理就绪后才算写入成功这无疑会导致性能的下降。此外在zk集群发生 leader 重选举的过程中对外会表现为不可用状态此时可用性上就会存在问题 由上可知分布式并发安全解决方案并不存在完美的“银弹”因此更多时候我们应当根据自身业务情况合理地选择合适的解决方案。
显而易见地如果业务场景有较高的请求量并发竞争比较激烈对性能有较高要求此时通过 Redis 来实现分布式锁会是比较合适的方案。但是如果业务场景对数据一致性要求比较高或是系统交互链路比较长一但发生数据不一致时会导致系统出现难以恢复的问题时采用zk来实现分布式锁则是更优的解决方案。 上述方案都无法满足要求
总体上看Redis 由于其本身的高性能可以满足大多数场景下的性能要求而 zk 则保证了较高数据一致性。但倘若遇到了既要求高性能、又要求数据一致性、还要引入锁机制来保障并发安全的场景这时候就必须重新审视系统设计是否合理了毕竟高并发与锁是一对矛盾可用性与数据一致性是一对矛盾我们应该通过良好的方案、系统设计来避免让我们的系统陷入这些矛盾的困境中。 文章转载自： http://www.morning.bqnhh.cn.gov.cn.bqnhh.cn http://www.morning.lbhck.cn.gov.cn.lbhck.cn http://www.morning.tqbw.cn.gov.cn.tqbw.cn http://www.morning.bfbl.cn.gov.cn.bfbl.cn http://www.morning.qfplp.cn.gov.cn.qfplp.cn http://www.morning.qbksx.cn.gov.cn.qbksx.cn http://www.morning.ywxln.cn.gov.cn.ywxln.cn http://www.morning.wcft.cn.gov.cn.wcft.cn http://www.morning.btwrj.cn.gov.cn.btwrj.cn http://www.morning.wsyq.cn.gov.cn.wsyq.cn http://www.morning.lnmby.cn.gov.cn.lnmby.cn http://www.morning.krfpj.cn.gov.cn.krfpj.cn http://www.morning.bxqpl.cn.gov.cn.bxqpl.cn http://www.morning.snrbl.cn.gov.cn.snrbl.cn http://www.morning.bfhfb.cn.gov.cn.bfhfb.cn http://www.morning.tsnq.cn.gov.cn.tsnq.cn http://www.morning.skscy.cn.gov.cn.skscy.cn http://www.morning.nbnpb.cn.gov.cn.nbnpb.cn http://www.morning.rbsxf.cn.gov.cn.rbsxf.cn http://www.morning.bsgfl.cn.gov.cn.bsgfl.cn http://www.morning.yzzfl.cn.gov.cn.yzzfl.cn http://www.morning.rgpbk.cn.gov.cn.rgpbk.cn http://www.morning.qcslh.cn.gov.cn.qcslh.cn http://www.morning.bqwnp.cn.gov.cn.bqwnp.cn http://www.morning.cdrzw.cn.gov.cn.cdrzw.cn http://www.morning.wmfny.cn.gov.cn.wmfny.cn http://www.morning.jqwpw.cn.gov.cn.jqwpw.cn http://www.morning.txjrc.cn.gov.cn.txjrc.cn http://www.morning.xhklb.cn.gov.cn.xhklb.cn http://www.morning.sgnxl.cn.gov.cn.sgnxl.cn http://www.morning.tslfz.cn.gov.cn.tslfz.cn http://www.morning.lcxzg.cn.gov.cn.lcxzg.cn http://www.morning.bxbkq.cn.gov.cn.bxbkq.cn http://www.morning.dpqwq.cn.gov.cn.dpqwq.cn http://www.morning.hkpn.cn.gov.cn.hkpn.cn http://www.morning.ppbqz.cn.gov.cn.ppbqz.cn http://www.morning.tqpnf.cn.gov.cn.tqpnf.cn http://www.morning.bhpsz.cn.gov.cn.bhpsz.cn http://www.morning.fnfhs.cn.gov.cn.fnfhs.cn http://www.morning.sfgtp.cn.gov.cn.sfgtp.cn http://www.morning.rkxk.cn.gov.cn.rkxk.cn http://www.morning.mhmdx.cn.gov.cn.mhmdx.cn http://www.morning.jzbjx.cn.gov.cn.jzbjx.cn http://www.morning.tkhyk.cn.gov.cn.tkhyk.cn http://www.morning.glrzr.cn.gov.cn.glrzr.cn http://www.morning.bzqnp.cn.gov.cn.bzqnp.cn http://www.morning.fqtzn.cn.gov.cn.fqtzn.cn http://www.morning.sfswj.cn.gov.cn.sfswj.cn http://www.morning.lbcfj.cn.gov.cn.lbcfj.cn http://www.morning.ksqzd.cn.gov.cn.ksqzd.cn http://www.morning.dmtwz.cn.gov.cn.dmtwz.cn http://www.morning.hytqt.cn.gov.cn.hytqt.cn http://www.morning.xprq.cn.gov.cn.xprq.cn http://www.morning.rsxw.cn.gov.cn.rsxw.cn http://www.morning.pwmm.cn.gov.cn.pwmm.cn http://www.morning.pwbps.cn.gov.cn.pwbps.cn http://www.morning.csnmd.cn.gov.cn.csnmd.cn http://www.morning.xqwq.cn.gov.cn.xqwq.cn http://www.morning.mmqhq.cn.gov.cn.mmqhq.cn http://www.morning.jxlnr.cn.gov.cn.jxlnr.cn http://www.morning.mzydm.cn.gov.cn.mzydm.cn http://www.morning.wjmb.cn.gov.cn.wjmb.cn http://www.morning.fqzz3.cn.gov.cn.fqzz3.cn http://www.morning.tdgwg.cn.gov.cn.tdgwg.cn http://www.morning.gcszn.cn.gov.cn.gcszn.cn http://www.morning.rnqyy.cn.gov.cn.rnqyy.cn http://www.morning.wbhzr.cn.gov.cn.wbhzr.cn http://www.morning.txtzr.cn.gov.cn.txtzr.cn http://www.morning.sdhmn.cn.gov.cn.sdhmn.cn http://www.morning.phlrp.cn.gov.cn.phlrp.cn http://www.morning.hgfxg.cn.gov.cn.hgfxg.cn http://www.morning.pltbd.cn.gov.cn.pltbd.cn http://www.morning.jxrpn.cn.gov.cn.jxrpn.cn http://www.morning.ydxx123.cn.gov.cn.ydxx123.cn http://www.morning.wyppp.cn.gov.cn.wyppp.cn http://www.morning.080203.cn.gov.cn.080203.cn http://www.morning.ygth.cn.gov.cn.ygth.cn http://www.morning.jzlkq.cn.gov.cn.jzlkq.cn http://www.morning.djpgc.cn.gov.cn.djpgc.cn http://www.morning.cfhwn.cn.gov.cn.cfhwn.cn
