背景

　　在工作中接到一个需求：对于访问页面的前x名用户分发A奖品，x+1名及以后的用户分发另外一种奖品。在J2EE的开发中，我们知道servlet是单实例多线程的，Spring的Controller类也一样，所以这里需要考虑多线程并发时如何判断该用户是否为前x名。一种办法是在代码中用内存控制，例如添加一个成员变量，创建一个方法，并在内部使用synchronized块对该变量加锁，每次调用这个方法时，来一个用户就先判断变量是否大于x，小于的话就对该变量+1，直到该变量超过x为止。但是因为我们的代码是部署在多台服务器上的，而在多台服务器上同步内存比较麻烦，所以这种方法只适用于一台服务器的情况。另一种方法就是在数据库级别加锁，因为我们的数据库只有一个节点，所以只要在这一个节点上加了锁就可以控制来访的用户了。

mysql锁机制简介

　　mysql提供了locking read机制，可以参考官方文档，一共有两种方式：SELECT ... FOR UPDATE和SELECT ... LOCK IN SHARE MODE。介绍它们之前，这里首先说一下X锁和S锁：

• 若事务 T 对数据对象 A 加了 X 锁，则 T 就可以对 A 进行读取以及更新。在 T 释放 A 上的 X 锁以前，其它事务不能对 A 加任何类型的锁，但可以使用普通select语句获取值，而这个值不能保证是最新的，因为事务 T 可能修改了 A 的值，而它还没有提交；

• 若事务 T 对数据对象 A 加了 S 锁，则 T 就可以对 A 进行读取，但不能进行更新。在 T 释放 A 上的 S 锁以前,其他事务可以再对 A 加 S 锁，但不能加 X 锁，从而可以读取 A ，但不能更新 A；

  SELECT ... FOR UPDATE是：

　　为选择的行添加排它锁(X锁)，保证查询到的数据是最新的数据，允许其它事务对该数据加上共享锁(S锁)，但不能修改，只有当前事务可以修改，其它事务需要等当前事务commit或rollback之后才可以修改加锁的行;

SELECT ... LOCK IN SHARE MODE是

　　为选择的行添加共享锁(S锁)，其它事务也可以对该行数据添加S锁，它保证了读取到的是最新的数据，并且不允许别人修改，但是自己也** 不一定 **能够修改，因为可能别的事务也对这个数据加了S锁;

实现

　　从上面对mysql锁的介绍可以看到，我的业务需要不仅读的时候要阻止别人读最新值，而且还可能要修改读取后的结果，因此这里使用SELECT ... FOR UPDATE语句来控制用户访问的排名最合适。
　　这里要注意一下，在mysql中用SELECT ... FOR UPDATE加锁，后面的WHERE条件是主键和非主键时有不同的加锁情况的，当WHERE后面是主键时，仅对行加锁，其它事务中可以对表的其他行进行增删改查，允许插入新的行；当WHERE后面的条件不是主键时，会锁全表，则其它事务不能对表的任意行进增删改的操作，插入新的行也不可以，只能查询。
　　首先在数据库创建一个简单的表，结构如下：

　　这个表中的每一行代表一个锁，也就是说下一次搞其它的活动，如果也需要对前x名进行控制，则插入一行记录用于代表一个锁。在java代码中，创建一个跟表映射的实体类LockBean，然后在DAO中添加两个方法，分别对应于查询和修改：

@Select(" select LOCK_KEY, LOCK_NUM, LOCK_DESC FROM LOCK_TABLE WHERE LOCK_KEY=#{lockKey} FOR UPDATE")
public LockBean findCurrentLock(int lockKey);@Update(" update LOCK_TABLE set LOCK_NUM = #{lockNum} where LOCK_KEY = #{lockKey}  ")
public void updateCurrentLock(LockBean lockBean);

　　最后，在service层中添加事务控制，保证这两个DAO的方法在一个事务里面执行。需要注意的是，SELECT ... FOR UPDATE语句必须要关闭自动提交，例如使用普通的JDBC来调用，则需要先调用 connection.setAutocommit(flase)关闭自动commit操作，然后在select和update之后，再调用connection.commit()提交事务。如果想要在Navicat或mysql workbench中测试locking read功能，则需要先执行set autocommit=0语句关闭自动提交，然后再进行操作。

优化

　　上面的方法对于每一次用户请求，都需要通过数据库级别的SELECT ... FOR UPDATE语句来加锁，可是往往前x名用户在总用户中所占的比例都是比较小的，毕竟大奖总是掌握在少数人手中嘛！如果每次都访问数据库，这样IO次数多了(同样也会导致网络请求次数增多，因为数据库只有一个节点)就会影响性能，所以我们在内存中再添加一个控制。在某个类中创建一个变量，用于判断前x名的奖品是否已经分发完毕：

public static volatile boolean isQueryNecessary = true;

　　顺便复习一下，要使得volatile变量提供理想的线程安全，必须同时满足以下两个条件：

1. 对变量的写操作不依赖于当前值
2. 该变量没有bao含在具有其他变量的不变式中

　　当变量声明为volatile后，所有线程对该对象的读取都会直接从主内存中获取，不会使用缓存的值，而在CPU缓存的一些值都会被标识为过期，从而完成线程对该对象的同步操作。具体介绍可见 Java 理论与实践: 正确使用 Volatile 变量.
　　回归正题，在service层的处理方法giveAward()中，伪代码如下：

if(true == isQueryNecessary) {// 如果isQueryNecessary为真，则查询数据库，注意这里可能需要等待有X锁的线程释放锁LockBean bean = dao.findCurrentLock(lockKey);/** 判断bean中的lockNum是否>=x*  true ：此时可能刚好等于x，也可能是在查询数据库时被别的线程抢先并更新了锁，*         即奖品别别人先抢完了，总之需要更新isQueryNecessary的值为false*         isQueryNecessary = false;*  false：lockNum++，*         dao.updateCurrentLock(bean); */
}if(false == isQueryNecessary) {// 再次判断是因为之前在查询数据库的时候有可能结果是lockNum >= x，// 导致isQueryNecessary的值被更新为false了// 总之这里处理x+1名以后的用户的逻辑logicForUserAfterX();
}

　　这里对isQueryNecessary判断了两次，主要是因为在多线程抢资源的情况下，变量的值可能会在等待过程中改变，所以采用单例模式中DCL的思想，双重判断，从而确保对每个用户请求正确分流。
　　通过这种优化后，对于单台服务器，顶多在第x个用户之后的部分请求(因为这些请求可能在抢第x个席位的过程中等待)会发生多于的数据库查询操作；而对于多台服务器，也只有部分的请求会执行多于的数据库查询，只要有一个请求在查询数据库之后发现已经不满足条件了就会把isQueryNecessary设为false，这台服务器后续的请求就不会再去查询数据库了，当全部的服务器上的isQueryNecessary都设为false之后，集群中后续的所有请求就都不再会查询数据库了，这样可以节省很多IO和网络操作。

redis实现方案

1. setnx方案

　　redis的 setnx 命令可以用来实现分布式锁的功能，因此可以把奖品数量放到redis中，例如系统加载时从DB获取到奖品总数为80，则SET AWARDNUM 80，接下来每个请求线程中用setnx命令加分布式锁（具体实现可以参考网上的方案，思路是给一个常量设置值，即setnx constant value，value为随机值，设置可以的过期时间，这样只有当前线程能释放该分布式锁，若没有及时释放也可以等待锁过期后重新尝试获取），获取到分布式锁后，先判断奖品库存是否<=0，如是则同步更新内存变量，避免下次再查询redis；如果>0则表示秒杀成功，然后对该奖品数量减一，并释放分布式锁即可。

2. MQ方案

　　该方案参考了这篇博文。redis有多种数据结构，例如链表，它可以作为一个MQ来使用，例如每个秒杀请求都放到队列中，再启动其它的线程去处理队列中前n个请求作为秒杀成功的处理。但是还有更简单的实现方案，例如系统初始化时从DB获取奖品数量为80，则初始化一个长度为80的list作为奖池，每个秒杀请求进来时使用LPOP或RPOP命令从list中抽取一个奖品，如果返回值为空，则说明奖池已经空了，否则表示秒杀成功。因为redis命令执行的时候都是单线程的原子操作，所以该方案的好处是实现简单且不需要用分布式锁，感觉分布式锁可能会更耗时间，因为即要加锁又要更新奖品数量，而这个方案只要读一次redis就可以了。