数据准备（构造秒杀队列）
秒杀数据
| seckill_id | name | number | create_time | version |
|---|---|---|---|---|
| 1000 | 秒杀iphone8 | 100 | 2018-05-10 15:31:53 | 100 |
秒杀执行代码：
//核心线程池大小=内核数
private static int corePoolSize = Runtime.getRuntime().availableProcessors();
//创建线程池 调整队列数 拒绝服务
private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(1000));
// 模拟1000用户，同时对商品进行秒杀操作(商品库存为100)
for(int i=0;i<1000;i++){
final long userId = i;
Runnable task = new Runnable() {
@Override
public void run() {
Result result = seckillService.startSeckil(killId, userId);
}
};
executor.execute(task);
}
单机并发秒杀处理
实现1 直接事务并发
@Transactional
public Result startSeckil(long seckillId,long userId) {
//校验库存,并发时库存结果并不准确
String nativeSql = "SELECT number FROM seckill WHERE seckill_id=?";
Object object = dynamicQuery.nativeQueryObject(nativeSql, new Object[]{seckillId});
Long number = ((Number) object).longValue();
if(number>0){
//扣库存
nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=?";
dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
//创建订单
dynamicQuery.save(new SuccessKilled(seckillId,userId));
return Result.ok(SeckillStatEnum.SUCCESS);
}else{
return Result.error(SeckillStatEnum.END);
}
}
秒杀结果：超卖，本次测试秒杀到104件商品
超卖原因：第一句查询到的结果实际是其他线程更新前的结果
实现2 Lock锁
private Lock lock = new ReentrantLock(true);//互斥锁，单例
@Transactional
public Result startSeckilLock(long seckillId, long userId) {
try {
lock.lock();
String nativeSql = "SELECT number FROM seckill WHERE seckill_id=?";
Object object = dynamicQuery.nativeQueryObject(nativeSql, new Object[]{seckillId});
Long number = ((Number) object).longValue();
if(number>0){
nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=?";
dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
//创建订单
dynamicQuery.save(new SuccessKilled(seckillId,userId));
}else{
return Result.error(SeckillStatEnum.END);
}
} catch (Exception e) {
e.printStackTrace();
}finally {
lock.unlock();
}
return Result.ok(SeckillStatEnum.SUCCESS);
}
秒杀结果：超卖，本次测试秒杀到101件商品
超卖原因：lock锁可以保证每次只有一个线程执行，但此处用法不对，lock.unlock()执行时，事务还未提交（事务注解会在方法执行完毕之后自动提交事务）
实现3 AOP+Lock锁
@Component
@Scope
@Aspect
@Order(1)//order越小越是最先执行，但更重要的是最先执行的最后结束
public class LockAspect {
/**
* service 默认是单例的，并发下lock只有一个实例
*/
private static Lock lock = new ReentrantLock(true);//互斥锁 参数默认false，不公平锁
//Service层切点
@Pointcut("@annotation(com.itstyle.seckill.common.aop.Servicelock)")
public void lockAspect() {
}
@Around("lockAspect()")
public Object around(ProceedingJoinPoint joinPoint) {
lock.lock();
Object obj = null;
try {
obj = joinPoint.proceed();
} catch (Throwable e) {
e.printStackTrace();
} finally{
lock.unlock();
}
return obj;
}
}
@Servicelock
@Transactional
public Result startSeckilAopLock(long seckillId, long userId) {
String nativeSql = "SELECT number FROM seckill WHERE seckill_id=?";
Object object = dynamicQuery.nativeQueryObject(nativeSql, new Object[]{seckillId});
Long number = ((Number) object).longValue();
if(number>0){
nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=?";
dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
//创建订单
dynamicQuery.save(new SuccessKilled(seckillId,userId));
}else{
return Result.error(SeckillStatEnum.END);
}
return Result.ok(SeckillStatEnum.SUCCESS);
}
秒杀结果：结果与预期一致，秒杀到100件商品
实现4 数据库悲观锁
@Transactional
public Result startSeckilDBPCC_ONE(long seckillId, long userId) {
//单用户抢购一件商品或者多件都没有问题
String nativeSql = "SELECT number FROM seckill WHERE seckill_id=? FOR UPDATE";
Object object = dynamicQuery.nativeQueryObject(nativeSql, new Object[]{seckillId});
Long number = ((Number) object).longValue();
if(number>0){
nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=?";
dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
//创建订单
dynamicQuery.save(new SuccessKilled(seckillId,userId));
return Result.ok(SeckillStatEnum.SUCCESS);
}else{
return Result.error(SeckillStatEnum.END);
}
}
秒杀结果：本次秒杀出9件商品，其余全部出现空指针异常
结果分析: 使用select ... for update 语句，有主键时会根据主键进行行锁，其他事务必须等待当前事务提交后才能执行
实现5：数据库UPDATE语句锁
@Transactional
public Result startSeckilDBPCC_TWO(long seckillId, long userId) {
String nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=? AND number>1";//UPDATE锁表
int count = dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
if(count>0){
dynamicQuery.save(new SuccessKilled(seckillId,userId));
return Result.ok(SeckillStatEnum.SUCCESS);
}else{
return Result.error(SeckillStatEnum.END);
}
}
秒杀结果：结果与预期一致，秒杀到100件商品
实现6：数据库乐观锁
@Transactional
public Result startSeckilDBOCC(long seckillId, long userId, long number) {
Seckill kill = seckillRepository.findOne(seckillId);
if(kill.getNumber()>=number){//剩余的数量应该要大于等于秒杀的数量
//乐观锁
String nativeSql = "UPDATE seckill SET number=number-?,version=version+1 WHERE seckill_id=? AND version = ?";
int count = dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{number,seckillId,kill.getVersion()});
if(count>0){
dynamicQuery.save(new SuccessKilled(seckillId,userId));
return Result.ok(SeckillStatEnum.SUCCESS);
}else{
return Result.error(SeckillStatEnum.END);
}
}else{
return Result.error(SeckillStatEnum.END);
}
}
秒杀结果：秒杀到100件商品，从用户ID来看基本是间隔几个用户成功一个
结果分析：每次同时并发时乐观锁会出现错误，排队靠前也很容易失败
实现7：进程内队列
/** 用于多线程间下单的队列，大小为100 */
static BlockingQueue<SuccessKilled> blockingQueue = new LinkedBlockingQueue<SuccessKilled>(100);
// 生产队列
public Result startQueue(long seckillId, long userId, long number) {
SuccessKilled kill = new SuccessKilled(seckillId,userId);
try {
Boolean flag = blockingQueue.offer(kill);//队列未满时，返回true
if(flag){
return Result.ok(SeckillStatEnum.SUCCESS);
}else{
return Result.error(SeckillStatEnum.END);
}
} catch (InterruptedException e) {
e.printStackTrace();
return Result.error(SeckillStatEnum.END);
}
}
//消费队列
public void run(ApplicationArguments var) throws Exception{
while(true){
//进程内队列
SuccessKilled kill = blockingQueue.take();
if(kill!=null){
seckillService.startSeckil(kill.getSeckillId(), kill.getUserId());
}
}
}
秒杀结果：秒杀到100件商品，与预期一致
结果分析：BlockingQueue为java.util.concurrent包中的工具类，它通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出。理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然
分布式（集群）秒杀处理
实现1：Rediss分布式锁
rediss分布式锁可使用redis的SET NX模式(只有key值不存在时才创建),每次进行秒杀前使用秒杀ID作为key存入redis，秒杀完成后删除此key值，从而实现分布式环境下的事务处理
//Rediss锁实现，也可以采用开源的redisson
public class RedisTool {
/**
* 尝试获取分布式锁
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @param expireTime 超期时间
* @return 是否获取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);//第三个参数为NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；第四个参数传的是PX，意思是我们要给这个key加一个过期的设置，具体时间由第五个参数决定
return "OK".equals(result);
}
/**
* 释放分布式锁
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @return 是否释放成功
*/
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
//首先获取锁对应的value值，检查是否与requestId相等，如果相等则删除锁（解锁）。那么为什么要使用Lua语言来实现呢？因为要确保上述操作是原子性的。
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
}
}
//秒杀代码类似单机秒杀-AOP+Lock锁，将本地锁换成redis分布式锁，另外增加finally代码释放锁（防止代码异常导致锁一致不释放）即可
说明：redis分布式锁速度较快，但redis无法知晓客户端连接断开，在网络故障导致锁释放请求未提交成功时，会导致redis锁一直等待（解决方案是对获取锁做超时等待时间做限制，但这样又可能因为事务执行时间过长导致并发结果异常）
实现2：zookeeper分布式锁
zookeeper分布式锁使用zookeeper临时节点实现，请求锁时在locker节点下创建zookeeper临时顺序节点，释放锁时删除此临时节点。
创建临时顺序节点后，需要调用getChildren(“locker”)来获取locker下面的所有子节点，如果发现自己在之前创建的子节点序号最小，那么就认为该客户端获取到了锁，否则监听上一个节点的删除事件等待获取锁，流程如下:
graph LR
开始(开始)-->C{创建Lock节点}
C--> |成功| 持有锁
C--> |失败| 设置watcher并等待
持有锁--> 删除Lock节点
删除Lock节点-.->设置watcher并等待
删除Lock节点-->结束(结束)
public class ZkLockUtil{
private static String address = "127.0.0.1:2181";
public static CuratorFramework client;
static{
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
client = CuratorFrameworkFactory.newClient(address, retryPolicy);
client.start();
}
private static InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
/**
* 私有的默认构造子，保证外界无法直接实例化
*/
private ZkLockUtil(){};
//获得锁
public static boolean acquire(long time, TimeUnit unit){
try {
return mutex.acquire(time,unit);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
//释放锁
public static void release(){
try {
mutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
本文参考
对应代码源码
网友评论