概述

幂等性原本是数学上的概念，即使公式：f(x)=f(f(x)) 能够成立的数学性质。用在编程领域，则意为对同一个系统，使用同样的条件，一次请求和重复的多次请求对系统资源的影响是一致的。

幂等性是分布式系统设计中十分重要的概念，具有这一性质的接口在设计时总是秉持这样的一种理念：调用接口发生异常并且重复尝试时，总是会造成系统所无法承受的损失，所以必须阻止这种现象的发生。

幂等有两个维度：一是空间维度上的幂等，即幂等对象的范围，是个人还是机构，是某一次交易还是某种类型的交易...二是时间维度上的幂等，即幂等的保证时间，是几秒、几分钟还是永久性的...

不同的需求，会有不一样的解决方案，难度和成本也不一样。

幂等性适用领域

试想这样的一种场景：在电商平台上支付后，因为网络原因导致系统提示你支付失败，于是你又重新付款了一次，等完成后检查网银发现被系统扣了两次款，这是一种什么样的体验？

造成上述问题的原因可能有很多，比如第一次付款时实际支付成功，但是信息返回时网络中断导致系统误判；又比如第一次付款的确失败了，但第二次付款时发生意外，导致支付请求被重复发送等等。在一次支付的过程中，每个环节都有可能会发生问题，我们要如何规避这类问题引发的分险？

幂等性是解决这类问题的方案之一，所以在电商，银行，互联网金融等对数据准确性要求很高的领域中，这一特性具有十分重要的地位。

幂等的常用思路

1. MVCC：####

多版本并发控制，乐观锁的一种实现，在数据更新时需要去比较持有数据的版本号，版本号不一致的操作无法成功。例如博客点赞次数自动+1的接口：

public boolean addCount(Long id, Long version);

update blogTable set count= count+1,version=version+1 where id=321 and version=123 

每一个version只有一次执行成功的机会，一旦失败必须重新获取。

2. 去重表：####

利用数据库表单的特性来实现幂等，常用的一个思路是在表上构建唯一性索引，保证某一类数据一旦执行完毕，后续同样的请求再也无法成功写入。

例子还是上述的博客点赞问题，要想防止一个人重复点赞，可以设计一张表，将博客id与用户id绑定建立唯一索引，每当用户点赞时就往表中写入一条数据，这样重复点赞的数据就无法写入。

3. TOKEN机制：####

这种机制就比较重要了，适用范围较广，有多种不同的实现方式。其核心思想是为每一次操作生成一个唯一性的凭证，也就是token。一个token在操作的每一个阶段只有一次执行权，一旦执行成功则保存执行结果。对重复的请求，返回同一个结果。

以电商平台为例子，电商平台上的订单id就是最适合的token。当用户下单时，会经历多个环节，比如生成订单，减库存，减优惠券等等。

每一个环节执行时都先检测一下该订单id是否已经执行过这一步骤，对未执行的请求，执行操作并缓存结果，而对已经执行过的id，则直接返回之前的执行结果，不做任何操作。这样可以在最大程度上避免操作的重复执行问题，缓存起来的执行结果也能用于事务的控制等。

总结

幂等性是分布式领域的一把利刃，每一个有志与分布式领域的程序员都应该熟悉它的设计思想。

分布式消息系统作为实现分布式系统可扩展、可伸缩性的关键组件，需要具有高吞吐量、高可用等特点。而谈到消息系统的设计，就回避不了两个问题：

1. 消息的顺序问题

2. 消息的重复问题

RocketMQ作为阿里开源的一款高性能、高吞吐量的消息中间件，它是怎样来解决这两个问题的？RocketMQ有哪些关键特性？其实现原理是怎样的？

关键特性及其实现原理

一、顺序消息

消息有序指的是可以按照消息的发送顺序来消费。例如：一笔订单产生了 3 条消息，分别是订单创建、订单付款、订单完成。消费时，要按照顺序依次消费才有意义。与此同时多笔订单之间又是可以并行消费的。首先来看如下示例：

假如生产者产生了2条消息：M1、M2，要保证这两条消息的顺序，应该怎样做？你脑中想到的可能是这样：

你可能会采用这种方式保证消息顺序

假定M1发送到S1，M2发送到S2，如果要保证M1先于M2被消费，那么需要M1到达消费端被消费后，通知S2，然后S2再将M2发送到消费端。

这个模型存在的问题是，如果M1和M2分别发送到两台Server上，就不能保证M1先达到MQ集群，也不能保证M1被先消费。换个角度看，如果M2先于M1达到MQ集群，甚至M2被消费后，M1才达到消费端，这时消息也就乱序了，说明以上模型是不能保证消息的顺序的。如何才能在MQ集群保证消息的顺序？一种简单的方式就是将M1、M2发送到同一个Server上：

保证消息顺序，你改进后的方法

这样可以保证M1先于M2到达MQServer（生产者等待M1发送成功后再发送M2），根据先达到先被消费的原则，M1会先于M2被消费，这样就保证了消息的顺序。

这个模型也仅仅是理论上可以保证消息的顺序，在实际场景中可能会遇到下面的问题：

网络延迟问题

只要将消息从一台服务器发往另一台服务器，就会存在网络延迟问题。如上图所示，如果发送M1耗时大于发送M2的耗时，那么M2就仍将被先消费，仍然不能保证消息的顺序。即使M1和M2同时到达消费端，由于不清楚消费端1和消费端2的负载情况，仍然有可能出现M2先于M1被消费的情况。

那如何解决这个问题？将M1和M2发往同一个消费者，且发送M1后，需要消费端响应成功后才能发送M2。

聪明的你可能已经想到另外的问题：如果M1被发送到消费端后，消费端1没有响应，那是继续发送M2呢，还是重新发送M1？一般为了保证消息一定被消费，肯定会选择重发M1到另外一个消费端2，就如下图所示。

保证消息顺序的正确姿势

这样的模型就严格保证消息的顺序，细心的你仍然会发现问题，消费端1没有响应Server时有两种情况，一种是M1确实没有到达(数据在网络传送中丢失)，另外一种消费端已经消费M1且已经发送响应消息，只是MQ Server端没有收到。如果是第二种情况，重发M1，就会造成M1被重复消费。也就引入了我们要说的第二个问题，消息重复问题，这个后文会详细讲解。

回过头来看消息顺序问题，严格的顺序消息非常容易理解，也可以通过文中所描述的方式来简单处理。总结起来，要实现严格的顺序消息，简单且可行的办法就是：

保证生产者 - MQServer - 消费者是一对一对一的关系

这样的设计虽然简单易行，但也会存在一些很严重的问题，比如：

1. 并行度就会成为消息系统的瓶颈（吞吐量不够）

2. 更多的异常处理，比如：只要消费端出现问题，就会导致整个处理流程阻塞，我们不得不花费更多的精力来解决阻塞的问题。

但我们的最终目标是要集群的高容错性和高吞吐量。这似乎是一对不可调和的矛盾，那么阿里是如何解决的？

世界上解决一个计算机问题最简单的方法：“恰好”不需要解决它！——沈询

有些问题，看起来很重要，但实际上我们可以通过合理的设计或者将问题分解来规避。如果硬要把时间花在解决问题本身，实际上不仅效率低下，而且也是一种浪费。从这个角度来看消息的顺序问题，我们可以得出两个结论：

1. 不关注乱序的应用实际大量存在

2. 队列无序并不意味着消息无序

所以从业务层面来保证消息的顺序而不仅仅是依赖于消息系统，是不是我们应该寻求的一种更合理的方式？

最后我们从源码角度分析RocketMQ怎么实现发送顺序消息的。

RocketMQ通过轮询所有队列的方式来确定消息被发送到哪一个队列（负载均衡策略）。比如下面的示例中，订单号相同的消息会被先后发送到同一个队列中：

在获取到路由信息以后，会根据MessageQueueSelector实现的算法来选择一个队列，同一个OrderId获取到的肯定是同一个队列。

二、消息重复

上面在解决消息顺序问题时，引入了一个新的问题，就是消息重复。那么RocketMQ是怎样解决消息重复的问题呢？还是“恰好”不解决。

造成消息重复的根本原因是：网络不可达。只要通过网络交换数据，就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？

1. 消费端处理消息的业务逻辑保持幂等性

2. 保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现

第1条很好理解，只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID，如果新到的消息ID已经在日志表中，那么就不再处理这条消息。

第1条解决方案，很明显应该在消费端实现，不属于消息系统要实现的功能。第2条可以消息系统实现，也可以业务端实现。正常情况下出现重复消息的概率其实很小，如果由消息系统来实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由业务端自己处理消息重复的问题，这也是RocketMQ不解决消息重复的问题的原因。

RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重。

三、事务消息

RocketMQ除了支持普通消息，顺序消息，另外还支持事务消息。首先讨论一下什么是事务消息以及支持事务消息的必要性。我们以一个转帐的场景为例来说明这个问题：Bob向Smith转账100块。

在单机环境下，执行事务的情况，大概是下面这个样子：

单机环境下转账事务示意图

当用户增长到一定程度，Bob和Smith的账户及余额信息已经不在同一台服务器上了，那么上面的流程就变成了这样：

集群环境下转账事务示意图

这时候你会发现，同样是一个转账的业务，在集群环境下，耗时居然成倍的增长，这显然是不能够接受的。那如何来规避这个问题？

大事务 = 小事务 + 异步

将大事务拆分成多个小事务异步执行。这样基本上能够将跨机事务的执行效率优化到与单机一致。转账的事务就可以分解成如下两个小事务：

小事务+异步消息

图中执行本地事务（Bob账户扣款）和发送异步消息应该保证同时成功或者同时失败，也就是扣款成功了，发送消息一定要成功，如果扣款失败了，就不能再发送消息。那问题是：我们是先扣款还是先发送消息呢？

首先看下先发送消息的情况，大致的示意图如下：

事务消息：先发送消息

存在的问题是：如果消息发送成功，但是扣款失败，消费端就会消费此消息，进而向Smith账户加钱。

先发消息不行，那就先扣款吧，大致的示意图如下：

事务消息-先扣款

存在的问题跟上面类似：如果扣款成功，发送消息失败，就会出现Bob扣钱了，但是Smith账户未加钱。

可能大家会有很多的方法来解决这个问题，比如：直接将发消息放到Bob扣款的事务中去，如果发送失败，抛出异常，事务回滚。这样的处理方式也符合“恰好”不需要解决的原则。

这里需要说明一下：如果使用Spring来管理事物的话，大可以将发送消息的逻辑放到本地事物中去，发送消息失败抛出异常，Spring捕捉到异常后就会回滚此事物，以此来保证本地事物与发送消息的原子性。

RocketMQ支持事务消息，下面来看看RocketMQ是怎样来实现的。

RocketMQ实现发送事务消息

RocketMQ第一阶段发送Prepared消息时，会拿到消息的地址，第二阶段执行本地事物，第三阶段通过第一阶段拿到的地址去访问消息，并修改消息的状态。

细心的你可能又发现问题了，如果确认消息发送失败了怎么办？RocketMQ会定期扫描消息集群中的事物消息，如果发现了Prepared消息，它会向消息发送端(生产者)确认，Bob的钱到底是减了还是没减呢？如果减了是回滚还是继续发送确认消息呢？RocketMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

那我们来看下RocketMQ源码，是如何处理事务消息的。客户端发送事务消息的部分（完整代码请查看：rocketmq-example工程下的com.alibaba.rocketmq.example.transaction.TransactionProducer）：

接着查看sendMessageInTransaction方法的源码，总共分为3个阶段：发送Prepared消息、执行本地事务、发送确认消息。

endTransaction方法会将请求发往broker(mq server)去更新事务消息的最终状态：

1. 根据sendResult找到Prepared消息 ，sendResult包含事务消息的ID

2. 根据localTransaction更新消息的最终状态

如果endTransaction方法执行失败，数据没有发送到broker，导致事务消息的 状态更新失败，broker会有回查线程定时（默认1分钟）扫描每个存储事务状态的表格文件，如果是已经提交或者回滚的消息直接跳过，如果是prepared状态则会向Producer发起CheckTransaction请求，Producer会调用DefaultMQProducerImpl.checkTransactionState()方法来处理broker的定时回调请求，而checkTransactionState会调用我们的事务设置的决断方法来决定是回滚事务还是继续执行，最后调用endTransactionOneway让broker来更新消息的最终状态。

再回到转账的例子，如果Bob的账户的余额已经减少，且消息已经发送成功，Smith端开始消费这条消息，这个时候就会出现消费失败和消费超时两个问题，解决超时问题的思路就是一直重试，直到消费端消费消息成功，整个过程中有可能会出现消息重复的问题，按照前面的思路解决即可。

消费事务消息

这样基本上可以解决消费端超时问题，但是如果消费失败怎么办？阿里提供给我们的解决方法是：人工解决。大家可以考虑一下，按照事务的流程，因为某种原因Smith加款失败，那么需要回滚整个流程。如果消息系统要实现这个回滚流程的话，系统复杂度将大大提升，且很容易出现Bug，估计出现Bug的概率会比消费失败的概率大很多。这也是RocketMQ目前暂时没有解决这个问题的原因，在设计实现消息系统时，我们需要衡量是否值得花这么大的代价来解决这样一个出现概率非常小的问题，这也是大家在解决疑难问题时需要多多思考的地方。

20160321补充：在3.2.6版本中移除了事务消息的实现，所以此版本不支持事务消息，具体情况请参考rocketmq的issues：

• https://github.com/alibaba/RocketMQ/issues/65

• https://github.com/alibaba/RocketMQ/issues/138

• https://github.com/alibaba/RocketMQ/issues/156

四、Producer如何发送消息

Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡，如下图所示：

producer发送消息负载均衡

首先分析一下RocketMQ的客户端发送消息的源码：

在整个应用生命周期内，生产者需要调用一次start方法来初始化，初始化主要完成的任务有：

1. 如果没有指定namesrv地址，将会自动寻址

2. 启动定时任务：更新namesrv地址、从namsrv更新topic路由信息、清理已经挂掉的broker、向所有broker发送心跳...

3. 启动负载均衡的服务

初始化完成后，开始发送消息，发送消息的主要代码如下：

代码中需要关注的两个方法tryToFindTopicPublishInfo和selectOneMessageQueue。前面说过在producer初始化时，会启动定时任务获取路由信息并更新到本地缓存，所以tryToFindTopicPublishInfo会首先从缓存中获取topic路由信息，如果没有获取到，则会自己去namesrv获取路由信息。selectOneMessageQueue方法通过轮询的方式，返回一个队列，以达到负载均衡的目的。

如果Producer发送消息失败，会自动重试，重试的策略：

1. 重试次数 < retryTimesWhenSendFailed（可配置）

2. 总的耗时（包含重试n次的耗时） < sendMsgTimeout（发送消息时传入的参数）

3. 同时满足上面两个条件后，Producer会选择另外一个队列发送消息

五、消息存储

RocketMQ的消息存储是由consume queue和commit log配合完成的。

1、Consume Queue

consume queue是消息的逻辑队列，相当于字典的目录，用来指定消息在物理文件commit log上的位置。

我们可以在配置中指定consumequeue与commitlog存储的目录

每个topic下的每个queue都有一个对应的consumequeue文件，比如：

Consume Queue文件组织，如图所示：

Consume Queue文件组织示意图

1. 根据topic和queueId来组织文件，图中TopicA有两个队列0,1，那么TopicA和QueueId=0组成一个ConsumeQueue，TopicA和QueueId=1组成另一个ConsumeQueue。

2. 按照消费端的GroupName来分组重试队列，如果消费端消费失败，消息将被发往重试队列中，比如图中的%RETRY%ConsumerGroupA。

3. 按照消费端的GroupName来分组死信队列，如果消费端消费失败，并重试指定次数后，仍然失败，则发往死信队列，比如图中的%DLQ%ConsumerGroupA。

死信队列（Dead Letter Queue）一般用于存放由于某种原因无法传递的消息，比如处理失败或者已经过期的消息。

Consume Queue中存储单元是一个20字节定长的二进制数据，顺序写顺序读，如下图所示：

consumequeue文件存储单元格式

1. CommitLog Offset是指这条消息在Commit Log文件中的实际偏移量

2. Size存储中消息的大小

3. Message Tag HashCode存储消息的Tag的哈希值：主要用于订阅时消息过滤（订阅时如果指定了Tag，会根据HashCode来快速查找到订阅的消息）

2、Commit Log

CommitLog：消息存放的物理文件，每台broker上的commitlog被本机所有的queue共享，不做任何区分。

文件的默认位置如下，仍然可通过配置文件修改：

${user.home} store${commitlog}${fileName}

CommitLog的消息存储单元长度不固定，文件顺序写，随机读。消息的存储结构如下表所示，按照编号顺序以及编号对应的内容依次存储。

Commit Log存储单元结构图

3、消息存储实现

消息存储实现，比较复杂，也值得大家深入了解，后面会单独成文来分析(目前正在收集素材)，这小节只以代码说明一下具体的流程。

4、消息的索引文件

如果一个消息包含key值的话，会使用IndexFile存储消息索引，文件的内容结构如图：

消息索引

索引文件主要用于根据key来查询消息的，流程主要是：

1. 根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目，例如图中所示 slotNum=5000000)

2. 根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)

3. 遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)

六、消息订阅

RocketMQ消息订阅有两种模式，一种是Push模式，即MQServer主动向消费端推送；另外一种是Pull模式，即消费端在需要时，主动到MQServer拉取。但在具体实现时，Push和Pull模式都是采用消费端主动拉取的方式。

首先看下消费端的负载均衡：

消费端负载均衡

消费端会通过RebalanceService线程，10秒钟做一次基于topic下的所有队列负载：

1. 遍历Consumer下的所有topic，然后根据topic订阅所有的消息

2. 获取同一topic和Consumer Group下的所有Consumer

3. 然后根据具体的分配策略来分配消费队列，分配的策略包含：平均分配、消费端配置等

如同上图所示：如果有 5 个队列，2 个 consumer，那么第一个 Consumer 消费 3 个队列，第二 consumer 消费 2 个队列。这里采用的就是平均分配策略，它类似于分页的过程，TOPIC下面的所有queue就是记录，Consumer的个数就相当于总的页数，那么每页有多少条记录，就类似于某个Consumer会消费哪些队列。

通过这样的策略来达到大体上的平均消费，这样的设计也可以很方面的水平扩展Consumer来提高消费能力。

消费端的Push模式是通过长轮询的模式来实现的，就如同下图：

Push模式示意图

Consumer端每隔一段时间主动向broker发送拉消息请求，broker在收到Pull请求后，如果有消息就立即返回数据，Consumer端收到返回的消息后，再回调消费者设置的Listener方法。如果broker在收到Pull请求时，消息队列里没有数据，broker端会阻塞请求直到有数据传递或超时才返回。

当然，Consumer端是通过一个线程将阻塞队列LinkedBlockingQueue<PullRequest>中的PullRequest发送到broker拉取消息，以防止Consumer一致被阻塞。而Broker端，在接收到Consumer的PullRequest时，如果发现没有消息，就会把PullRequest扔到ConcurrentHashMap中缓存起来。broker在启动时，会启动一个线程不停的从ConcurrentHashMap取出PullRequest检查，直到有数据返回。

七、RocketMQ的其他特性

前面的6个特性都是基本上都是点到为止，想要深入了解，还需要大家多多查看源码，多多在实际中运用。当然除了已经提到的特性外，RocketMQ还支持：

1. 定时消息

2. 消息的刷盘策略

3. 主动同步策略：同步双写、异步复制

4. 海量消息堆积能力

5. 高效通信

6. .......

其中涉及到的很多设计思路和解决方法都值得我们深入研究：

1. 消息的存储设计：既要满足海量消息的堆积能力，又要满足极快的查询效率，还要保证写入的效率。

2. 高效的通信组件设计：高吞吐量，毫秒级的消息投递能力都离不开高效的通信。

3. .......

RocketMQ最佳实践

一、Producer最佳实践

1. 一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。

2. 每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

   消息发送成功或者失败，要打印消息日志，务必要打印 sendresult 和 key 字段。

3. 对于消息不可丢失应用，务必要有消息重发机制。例如：消息发送失败，存储到数据库，能有定时程序尝试重发或者人工触发重发。

4. 某些应用如果不关注消息是否发送成功，请直接使用sendOneWay方法发送消息。

二、Consumer最佳实践

1. 消费过程要做到幂等（即消费端去重）

2. 尽量使用批量方式消费方式，可以很大程度上提高消费吞吐量。

3. 优化每条消息消费过程

三、其他配置

线上应该关闭autoCreateTopicEnable，即在配置文件中将其设置为false。

RocketMQ在发送消息时，会首先获取路由信息。如果是新的消息，由于MQServer上面还没有创建对应的Topic，这个时候，如果上面的配置打开的话，会返回默认TOPIC的（RocketMQ会在每台broker上面创建名为TBW102的TOPIC）路由信息，然后Producer会选择一台Broker发送消息，选中的broker在存储消息时，发现消息的topic还没有创建，就会自动创建topic。后果就是：以后所有该TOPIC的消息，都将发送到这台broker上，达不到负载均衡的目的。

所以基于目前RocketMQ的设计，建议关闭自动创建TOPIC的功能，然后根据消息量的大小，手动创建TOPIC。

RocketMQ设计相关

RocketMQ的设计假定：

• 每台PC机器都可能宕机不可服务

• 任意集群都有可能处理能力不足

• 最坏的情况一定会发生

• 内网环境需要低延迟来提供最佳用户体验

RocketMQ的关键设计：

• 分布式集群化

• 强数据安全

• 海量数据堆积

• 毫秒级投递延迟（推拉模式）

这是RocketMQ在设计时的假定前提以及需要到达的效果。我想这些假定适用于所有的系统设计。随着我们系统的服务的增多，每位开发者都要注意自己的程序是否存在单点故障，如果挂了应该怎么恢复、能不能很好的水平扩展、对外的接口是否足够高效、自己管理的数据是否足够安全...... 多多规范自己的设计，才能开发出高效健壮的程序。

参考资料

• RocketMQ用户指南

  https://pan.baidu.com/s/1kTWsE8J

• RocketMQ原理简介

  https://pan.baidu.com/s/1bogcpgN

• RocketMQ最佳实践

  https://pan.baidu.com/s/1kTXE4PD

• 阿里分布式开放消息服务(ONS)原理与实践2

  http://v.youku.com/v_show/id_XODY4ODE3OTY0.html?from=s1.8-1-1.2

• 阿里分布式开放消息服务(ONS)原理与实践3

  http://v.youku.com/v_show/id_XODY5ODcxNjI0.html?from=s1.8-1-1.2

• RocketMQ原理解析

  http://blog.csdn.net/column/details/learningrocketmq.html

水平有限，难免疏漏，如有问题请留言。

经作者同意授权转载

来源：简书

作者：CHEN川

Java高效并发之乐观锁悲观锁、（互斥同步、非互斥同步）

乐观锁和悲观锁

首先我们理解下两种不同思路的锁，乐观锁和悲观锁。
这两种锁机制，是在多用户环境并发控制的两种所机制。下面看百度百科对乐观锁和悲观锁两种锁机制的定义：

乐观锁（ Optimistic Locking ） 相对悲观锁而言，乐观锁机制采取了更加宽松的加锁机制。悲观锁大多数情况下依靠数据库的锁机制实现，以保证操作最大程度的独占性。但随之而来的就是数据库性能的大量开销，特别是对长事务而言，这样的开销往往无法承受。而乐观锁机制在一定程度上解决了这个问题。乐观锁，大多是基于数据版本（ Version ）记录机制实现。何谓数据版本？即为数据增加一个版本标识，在基于数据库表的版本解决方案中，一般是通过为数据库表增加一个 “version” 字段来实现。读取出数据时，将此版本号一同读出，之后更新时，对此版本号加一。此时，将提交数据的版本数据与数据库表对应记录的当前版本信息进行比对，如果提交的数据版本号大于数据库表当前版本号，则予以更新，否则认为是过期数据。
悲观锁（Pessimistic Lock），正如其名，具有强烈的独占和排他特性。它指的是对数据被外界（包括本系统当前的其他事务，以及来自外部系统的事务处理）修改持保守态度，因此，在整个数据处理过程中，将数据处于锁定状态。悲观锁的实现，往往依靠数据库提供的锁机制（也只有数据库层提供的锁机制才能真正保证数据访问的排他性，否则，即使在本系统中实现了加锁机制，也无法保证外部系统不会修改数据）。

简而言之：
悲观锁：假定会发生并发冲突，屏蔽一切可能违反数据完整性的操作。[1]
乐观锁：假设不会发生并发冲突，只在提交操作时检查是否违反数据完整性。[1] 乐观锁不能解决脏读的问题。
Java中的乐观锁和悲观锁：我们都知道，cpu是时分复用的，也就是把cpu的时间片，分配给不同的thread/process轮流执行，时间片与时间片之间，需要进行cpu切换，也就是会发生进程的切换。切换涉及到清空寄存器，缓存数据。然后重新加载新的thread所需数据。当一个线程被挂起时，加入到阻塞队列，在一定的时间或条件下，在通过notify()，notifyAll()唤醒回来。在某个资源不可用的时候，就将cpu让出，把当前等待线程切换为阻塞状态。等到资源(比如一个共享数据）可用了，那么就将线程唤醒，让他进入runnable状态等待cpu调度。这就是典型的悲观锁的实现。独占锁是一种悲观锁，synchronized就是一种独占锁，它假设最坏的情况，并且只有在确保其它线程不会造成干扰的情况下执行，会导致其它所有需要锁的线程挂起，等待持有锁的线程释放锁。
但是，由于在进程挂起和恢复执行过程中存在着很大的开销。当一个线程正在等待锁时，它不能做任何事，所以悲观锁有很大的缺点。举个例子，如果一个线程需要某个资源，但是这个资源的占用时间很短，当线程第一次抢占这个资源时，可能这个资源被占用，如果此时挂起这个线程，可能立刻就发现资源可用，然后又需要花费很长的时间重新抢占锁，时间代价就会非常的高。
所以就有了乐观锁的概念，他的核心思路就是，每次不加锁而是假设没有冲突而去完成某项操作，如果因为冲突失败就重试，直到成功为止。在上面的例子中，某个线程可以不让出cpu,而是一直while循环，如果失败就重试，直到成功为止。所以，当数据争用不严重时，乐观锁效果更好。比如CAS就是一种乐观锁思想的应用。
JDK1.5中引入了底层的支持，在int、long和对象的引用等类型上都公开了CAS的操作，并且JVM把它们编译为底层硬件提供的最有效的方法，在运行CAS的平台上，运行时把它们编译为相应的机器指令。在java.util.concurrent.atomic包下面的所有的原子变量类型中，比如AtomicInteger，都使用了这些底层的JVM支持为数字类型的引用类型提供一种高效的CAS操作。
在CAS操作中，会出现ABA问题。就是如果V的值先由A变成B，再由B变成A，那么仍然认为是发生了变化，并需要重新执行算法中的步骤。有简单的解决方案：不是更新某个引用的值，而是更新两个值，包括一个引用和一个版本号，即使这个值由A变为B，然后为变为A，版本号也是不同的。AtomicStampedReference和AtomicMarkableReference支持在两个变量上执行原子的条件更新。AtomicStampedReference更新一个“对象-引用”二元组，通过在引用上加上“版本号”，从而避免ABA问题，AtomicMarkableReference将更新一个“对象引用-布尔值”的二元组。

引用《深入理解Java虚拟机第二版》中原文：

13.2.2　线程安全的实现方法
了解了什么是线程安全之后，紧接着的一个问题就是我们应该如何实现线程安全，这听
起来似乎是一件由代码如何编写来决定的事情，确实，如何实现线程安全与代码编写有很大
的关系，但虚拟机提供的同步和锁机制也起到了非常重要的作用。本节中，代码编写如何实
现线程安全和虚拟机如何实现同步与锁这两者都会有所涉及，相对而言更偏重后者一些，只
要读者了解了虚拟机线程安全手段的运作过程，自己去思考代码如何编写并不是一件困难的
事情。
1.互斥同步
互斥同步（Mutual Exclusion＆Synchronization）是常见的一种并发正确性保障手段。同步
是指在多个线程并发访问共享数据时，保证共享数据在同一个时刻只被一个（或者是一些，
使用信号量的时候）线程使用。而互斥是实现同步的一种手段，临界区（Critical
Section）、互斥量（Mutex）和信号量（Semaphore）都是主要的互斥实现方式。因此，在这
4个字里面，互斥是因，同步是果；互斥是方法，同步是目的。
在Java中，最基本的互斥同步手段就是synchronized关键字，synchronized关键字经过编译
之后，会在同步块的前后分别形成monitorenter和monitorexit这两个字节码指令，这两个字节
码都需要一个reference类型的参数来指明要锁定和解锁的对象。如果Java程序中的
synchronized明确指定了对象参数，那就是这个对象的reference；如果没有明确指定，那就根
据synchronized修饰的是实例方法还是类方法，去取对应的对象实例或Class对象来作为锁对
象。
根据虚拟机规范的要求，在执行monitorenter指令时，首先要尝试获取对象的锁。如果这
个对象没被锁定，或者当前线程已经拥有了那个对象的锁，把锁的计数器加1，相应的，在
执行monitorexit指令时会将锁计数器减1，当计数器为0时，锁就被释放。如果获取对象锁失
败，那当前线程就要阻塞等待，直到对象锁被另外一个线程释放为止。
在虚拟机规范对monitorenter和monitorexit的行为描述中，有两点是需要特别注意的。首
先，synchronized同步块对同一条线程来说是可重入的，不会出现自己把自己锁死的问题。其
次，同步块在已进入的线程执行完之前，会阻塞后面其他线程的进入。第12章讲过，Java的
线程是映射到操作系统的原生线程之上的，如果要阻塞或唤醒一个线程，都需要操作系统来
帮忙完成，这就需要从用户态转换到核心态中，因此状态转换需要耗费很多的处理器时间。
对于代码简单的同步块（如被synchronized修饰的getter（）或setter（）方法），状态转换消
耗的时间有可能比用户代码执行的时间还要长。所以synchronized是Java语言中一个重量级
（Heavyweight）的操作，有经验的程序员都会在确实必要的情况下才使用这种操作。而虚拟
机本身也会进行一些优化，譬如在通知操作系统阻塞线程之前加入一段自旋等待过程，避免
频繁地切入到核心态之中。
除了synchronized之外，我们还可以使用java.util.concurrent（下文称J.U.C）包中的重入锁
（ReentrantLock）来实现同步，在基本用法上，ReentrantLock与synchronized很相似，他们都
具备一样的线程重入特性，只是代码写法上有点区别，一个表现为API层面的互斥锁
（lock（）和unlock（）方法配合try/finally语句块来完成），另一个表现为原生语法层面的互
斥锁。不过，相比synchronized,ReentrantLock增加了一些高级功能，主要有以下3项：等待可
中断、可实现公平锁，以及锁可以绑定多个条件。
等待可中断是指当持有锁的线程长期不释放锁的时候，正在等待的线程可以选择放弃等
待，改为处理其他事情，可中断特性对处理执行时间非常长的同步块很有帮助。
公平锁是指多个线程在等待同一个锁时，必须按照申请锁的时间顺序来依次获得锁；而
非公平锁则不保证这一点，在锁被释放时，任何一个等待锁的线程都有机会获得锁。
synchronized中的锁是非公平的，ReentrantLock默认情况下也是非公平的，但可以通过带布尔
值的构造函数要求使用公平锁。
锁绑定多个条件是指一个ReentrantLock对象可以同时绑定多个Condition对象，而在
synchronized中，锁对象的wait（）和notify（）或notifyAll（）方法可以实现一个隐含的条
件，如果要和多于一个的条件关联的时候，就不得不额外地添加一个锁，而ReentrantLock则
无须这样做，只需要多次调用newCondition（）方法即可。
如果需要使用上述功能，选用ReentrantLock是一个很好的选择，那如果是基于性能考虑
呢？关于synchronized和ReentrantLock的性能问题，Brian Goetz对这两种锁在JDK 1.5与单核处
理器，以及JDK 1.5与双Xeon处理器环境下做了一组吞吐量对比的实验 [1] ，实验结果如图13-1
和图13-2所示。
图　13-1　JDK 1.5、单核处理器下两种锁的吞吐量对比
从图13-1和图13-2可以看出，多线程环境下synchronized的吞吐量下降得非常严重，而
ReentrantLock则能基本保持在同一个比较稳定的水平上。与其说ReentrantLock性能好，还不
如说synchronized还有非常大的优化余地。后续的技术发展也证明了这一点，JDK 1.6中加入
了很多针对锁的优化措施（13.3节我们就会讲解这些优化措施），JDK 1.6发布之后，人们就
发现synchronized与ReentrantLock的性能基本上是完全持平了。因此，如果读者的程序是使用
JDK 1.6或以上部署的话，性能因素就不再是选择ReentrantLock的理由了，虚拟机在未来的性
能改进中肯定也会更加偏向于原生的synchronized，所以还是提倡在synchronized能实现需求
的情况下，优先考虑使用synchronized来进行同步。
图　13-2　JDK 1.5、双Xeon处理器下两种锁的吞吐量对比
2.非阻塞同步
互斥同步最主要的问题就是进行线程阻塞和唤醒所带来的性能问题，因此这种同步也称
为阻塞同步（Blocking Synchronization）。从处理问题的方式上说，互斥同步属于一种悲观的
并发策略，总是认为只要不去做正确的同步措施（例如加锁），那就肯定会出现问题，无论
共享数据是否真的会出现竞争，它都要进行加锁（这里讨论的是概念模型，实际上虚拟机会
优化掉很大一部分不必要的加锁）、用户态核心态转换、维护锁计数器和检查是否有被阻塞
的线程需要唤醒等操作。随着硬件指令集的发展，我们有了另外一个选择：基于冲突检测的
乐观并发策略，通俗地说，就是先进行操作，如果没有其他线程争用共享数据，那操作就成
功了；如果共享数据有争用，产生了冲突，那就再采取其他的补偿措施（最常见的补偿措施
就是不断地重试，直到成功为止），这种乐观的并发策略的许多实现都不需要把线程挂起，
因此这种同步操作称为非阻塞同步（Non-Blocking Synchronization）。
为什么笔者说使用乐观并发策略需要“硬件指令集的发展”才能进行呢？因为我们需要操
作和冲突检测这两个步骤具备原子性，靠什么来保证呢？如果这里再使用互斥同步来保证就
失去意义了，所以我们只能靠硬件来完成这件事情，硬件保证一个从语义上看起来需要多次
操作的行为只通过一条处理器指令就能完成，

这类指令常用的有：
测试并设置（Test-and-Set）。
获取并增加（Fetch-and-Increment）。
交换（Swap）。
比较并交换（Compare-and-Swap，下文称CAS）。
加载链接/条件存储（Load-Linked/Store-Conditional，下文称LL/SC）。

其中，前面的3条是20世纪就已经存在于大多数指令集之中的处理器指令，后面的两条
是现代处理器新增的，而且这两条指令的目的和功能是类似的。在IA64、x86指令集中有
cmpxchg指令完成CAS功能，在sparc-TSO也有casa指令实现，而在ARM和PowerPC架构下，
则需要使用一对ldrex/strex指令来完成LL/SC的功能。
CAS指令需要有3个操作数，分别是内存位置（在Java中可以简单理解为变量的内存地
址，用V表示）、旧的预期值（用A表示）和新值（用B表示）。CAS指令执行时，当且仅当
V符合旧预期值A时，处理器用新值B更新V的值，否则它就不执行更新，但是无论是否更新
了V的值，都会返回V的旧值，上述的处理过程是一个原子操作。
在JDK  1.5之后，Java程序中才可以使用CAS操作，该操作由sun.misc.Unsafe类里面的
compareAndSwapInt（）和compareAndSwapLong（）等几个方法包装提供，虚拟机在内部对
这些方法做了特殊处理，即时编译出来的结果就是一条平台相关的处理器CAS指令，没有方
法调用的过程，或者可以认为是无条件内联进去了 [2] 。
由于Unsafe类不是提供给用户程序调用的类（Unsafe.getUnsafe（）的代码中限制了只有
启动类加载器（Bootstrap  ClassLoader）加载的Class才能访问它），因此，如果不采用反射
手段，我们只能通过其他的Java  API来间接使用它，如J.U.C包里面的整数原子类，其中的
compareAndSet（）和getAndIncrement（）等方法都使用了Unsafe类的CAS操作。
我们不妨拿一段在第12章中没有解决的问题代码来看看如何使用CAS操作来避免阻塞同
步，代码如代码清单12-1所示。我们曾经通过这段20个线程自增10000次的代码来证明
volatile变量不具备原子性，那么如何才能让它具备原子性呢？把“race++”操作或increase（）
方法用同步块包裹起来当然是一个办法，但是如果改成如代码清单13-4所示的代码，那效率
将会提高许多。
代码清单13-4　Atomic的原子自增运算
/**
*Atomic变量自增运算测试
*
*@author
*/
public class AtomicTest{
public static AtomicInteger race=new AtomicInteger（0）；
public static void increase（）{
race.incrementAndGet（）；
}
private static final int THREADS_COUNT=20；
public static void main（String[]args）throws Exception{
Thread[]threads=new Thread[THREADS_COUNT]；
for（int i=0；i＜THREADS_COUNT；i++）{
threads[i]=new Thread（new Runnable（）{
@Override
public void run（）{
for（int i=0；i＜10000；i++）{
increase（）；
}
}
}）；
threads[i].start（）；
}
while（Thread.activeCount（）＞1）
Thread.yield（）；
System.out.println（race）；
}
}
运行结果如下：
200000
使用AtomicInteger代替int后，程序输出了正确的结果，一切都要归功于
incrementAndGet（）方法的原子性。它的实现其实非常简单，如代码清单13-5所示。
代码清单13-5　incrementAndGet（）方法的JDK源码
/**
*Atomically increment by one the current value.
*@return the updated value
*/
public final int incrementAndGet（）{
for（；）{
int current=get（）；
int next=current+1；
if（compareAndSet（current,next））
return next；
}
}
incrementAndGet（）方法在一个无限循环中，不断尝试将一个比当前值大1的新值赋给
自己。如果失败了，那说明在执行“获取-设置”操作的时候值已经有了修改，于是再次循环
进行下一次操作，直到设置成功为止。
尽管CAS看起来很美，但显然这种操作无法涵盖互斥同步的所有使用场景，并且CAS从
语义上来说并不是完美的，存在这样的一个逻辑漏洞：如果一个变量V初次读取的时候是A
值，并且在准备赋值的时候检查到它仍然为A值，那我们就能说它的值没有被其他线程改变
过了吗？如果在这段期间它的值曾经被改成了B，后来又被改回为A，那CAS操作就会误认
为它从来没有被改变过。这个漏洞称为CAS操作的“ABA”问题。J.U.C包为了解决这个问题，
提供了一个带有标记的原子引用类“AtomicStampedReference”，它可以通过控制变量值的版本
来保证CAS的正确性。不过目前来说这个类比较“鸡肋”，大部分情况下ABA问题不会影响程
序并发的正确性，如果需要解决ABA问题，改用传统的互斥同步可能会比原子类更高效。

MySQL 乐观锁与悲观锁

FX_SKY

CREATE TABLE `tb_product_stock` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增ID',
  `product_id` bigint(32) NOT NULL COMMENT '商品ID',
  `number` INT(8) NOT NULL DEFAULT 0 COMMENT '库存数量',
  `create_time` DATETIME NOT NULL COMMENT '创建时间',
  `modify_time` DATETIME NOT NULL COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `index_pid` (`product_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='商品库存表';

class ProductStock {
    private Long productId; //商品id
    private Integer number; //库存量

    public Long getProductId() {
        return productId;
    }

    public void setProductId(Long productId) {
        this.productId = productId;
    }

    public Integer getNumber() {
        return number;
    }

    public void setNumber(Integer number) {
        this.number = number;
    }
}

    /**
     * 更新库存(不考虑并发)
     * @param productId
     * @return
     */
    public boolean updateStockRaw(Long productId){
        ProductStock product = query("SELECT * FROM tb_product_stock WHERE product_id=#{productId}", productId);
        if (product.getNumber() > 0) {
            int updateCnt = update("UPDATE tb_product_stock SET number=number-1 WHERE product_id=#{productId}", productId);
            if(updateCnt > 0){    //更新库存成功
                return true;
            }
        }
        return false;
    }

/**
     * 更新库存(使用悲观锁)
     * @param productId
     * @return
     */
    public boolean updateStock(Long productId){
        //先锁定商品库存记录
        ProductStock product = query("SELECT * FROM tb_product_stock WHERE product_id=#{productId} FOR UPDATE", productId);
        if (product.getNumber() > 0) {
            int updateCnt = update("UPDATE tb_product_stock SET number=number-1 WHERE product_id=#{productId}", productId);
            if(updateCnt > 0){    //更新库存成功
                return true;
            }
        }
        return false;
    }

    /**
     * 下单减库存
     * @param productId
     * @return
     */
    public boolean updateStock(Long productId){
        int updateCnt = 0;
        while (updateCnt == 0) {
            ProductStock product = query("SELECT * FROM tb_product_stock WHERE product_id=#{productId}", productId);
            if (product.getNumber() > 0) {
                updateCnt = update("UPDATE tb_product_stock SET number=number-1 WHERE product_id=#{productId} AND number=#{number}", productId, product.getNumber());
                if(updateCnt > 0){    //更新库存成功
                    return true;
                }
            } else {    //卖完啦
                return false;
            }
        }
        return false;
    }