利用redis实现分布式锁,快速解决高并发时的线程安全问题

实际工作中,经常会遇到多线程并发时的类似抢购的功能,本篇描述一个简单的redis分布式锁" title="redis分布式锁">redis分布式锁实现的多线程抢票功能。

直接上代码。首先按照慣例,给出一个错误的示范:

我们可以看看,当20个线程一起来抢10张票的时候,会发生什么事。

package com.tiger.utils;

public class TestMutilThread {

// 总票量

public static int count = 10;

public static void main(String[] args) {

statrtMulti();

}

public static void statrtMulti() {

for (int i = 1; i <= 20; i++) {

TicketRunnable tickrunner = new TicketRunnable();

Thread thread = new Thread(tickrunner, "Thread No: " + i);

thread.start();

}

}

public static class TicketRunnable implements Runnable {

@Override

public void run() {

System.out.println(Thread.currentThread().getName() + " start "

+ count);

// TODO Auto-generated method stub

// logger.info(Thread.currentThread().getName()

// + " really start" + count);

if (count <= 0) {

System.out.println(Thread.currentThread().getName()

+ " ticket sold out ! No tickets remained!" + count);

return;

} else {

count = count - 1;

System.out.println(Thread.currentThread().getName()

+ " bought a ticket,now remaining :" + (count));

}

}

}

}

测试结果,从结果可以看到,票数在不同的线程中已经出现混乱。

Thread No: 2 start 10

Thread No: 6 start 10

Thread No: 4 start 10

Thread No: 5 start 10

Thread No: 3 start 10

Thread No: 9 start 6

Thread No: 1 start 10

Thread No: 1 bought a ticket,now remaining :3

Thread No: 9 bought a ticket,now remaining :4

Thread No: 3 bought a ticket,now remaining :5

Thread No: 12 start 3

Thread No: 5 bought a ticket,now remaining :6

Thread No: 4 bought a ticket,now remaining :7

Thread No: 8 start 7

Thread No: 7 start 8

Thread No: 12 bought a ticket,now remaining :1

Thread No: 14 start 0

Thread No: 6 bought a ticket,now remaining :8

Thread No: 16 start 0

Thread No: 2 bought a ticket,now remaining :9

Thread No: 16 ticket sold out ! No tickets remained!0

Thread No: 14 ticket sold out ! No tickets remained!0

Thread No: 18 start 0

Thread No: 18 ticket sold out ! No tickets remained!0

Thread No: 7 bought a ticket,now remaining :0

Thread No: 15 start 0

Thread No: 8 bought a ticket,now remaining :1

Thread No: 13 start 2

Thread No: 19 start 0

Thread No: 11 start 3

Thread No: 11 ticket sold out ! No tickets remained!0

Thread No: 10 start 3

Thread No: 10 ticket sold out ! No tickets remained!0

Thread No: 19 ticket sold out ! No tickets remained!0

Thread No: 13 ticket sold out ! No tickets remained!0

Thread No: 20 start 0

Thread No: 20 ticket sold out ! No tickets remained!0

Thread No: 15 ticket sold out ! No tickets remained!0

Thread No: 17 start 0

Thread No: 17 ticket sold out ! No tickets remained!0

为了解决多线程时出现的混乱问题,这里給出真正的测试类!!!

真正的测试类,这里启动20个线程,来抢10张票。

RedisTemplate 是用来实现redis操作的,由spring进行集成。这里是使用到了RedisTemplate,所以我以构造器的形式在外部将RedisTemplate传入到测试类中。

MultiTestLock 是用来实现加锁的工具类。

总票数使用volatile关键字,实现多线程时变量在系统内存中的可见性,这点可以去了解下volatile关键字的作用。

TicketRunnable用于模拟抢票功能。

其中由于lock与unlock之间存在if判断,为保证线程安全,这里使用synchronized来保证。

测试类:

package com.tiger.utils;

import java.io.Serializable;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.data.redis.core.RedisTemplate;

public class MultiConsumer {

Logger logger=LoggerFactory.getLogger(MultiTestLock.class);

private RedisTemplate<Serializable, Serializable> redisTemplate;

public MultiTestLock lock;

//总票量

public volatile static int count = 10;

public void statrtMulti() {

lock = new MultiTestLock(redisTemplate);

for (int i = 1; i <= 20; i++) {

TicketRunnable tickrunner = new TicketRunnable();

Thread thread = new Thread(tickrunner, "Thread No: " + i);

thread.start();

}

}

public class TicketRunnable implements Runnable {

@Override

public void run() {

logger.info(Thread.currentThread().getName() + " start "

+ count);

// TODO Auto-generated method stub

if (count > 0) {

// logger.info(Thread.currentThread().getName()

// + " really start" + count);

lock.lock();

synchronized (this) {

if(count<=0){

logger.info(Thread.currentThread().getName()

+ " ticket sold out ! No tickets remained!" + count);

lock.unlock();

return;

}else{

count=count-1;

logger.info(Thread.currentThread().getName()

+ " bought a ticket,now remaining :" + (count));

}

}

lock.unlock();

}else{

logger.info(Thread.currentThread().getName()

+ " ticket sold out !" + count);

}

}

}

public RedisTemplate<Serializable, Serializable> getRedisTemplate() {

return redisTemplate;

}

public void setRedisTemplate(

RedisTemplate<Serializable, Serializable> redisTemplate) {

this.redisTemplate = redisTemplate;

}

public MultiConsumer(RedisTemplate<Serializable, Serializable> redisTemplate) {

super();

this.redisTemplate = redisTemplate;

}

}

Lock工具类:

我们知道为保证线程安全,程序中执行的操作必须时原子的。redis后续的版本中可以使用set key同时设置expire超时时间。

想起上次去 电信翼支付 面试时,面试官问过一个问题:分布式锁如何防止死锁,问题关键在于我们在分布式中进行加锁操作时成功了,但是后续业务操作完毕执行解锁时出现失败。导致分布式锁无法释放。出现死锁,后续的加锁无法正常进行。所以这里设置expire超时时间的目的就是防止出现解锁失败的情况,这样,即使解锁失败了,分布式锁依然会在超时时间过了之后自动释放。

具体在代码中也有注释,也可以作为参考。

package com.tiger.utils;

import java.io.Serializable;

import java.util.Arrays;

import java.util.Collections;

import java.util.HashMap;

import java.util.Iterator;

import java.util.List;

import java.util.Random;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import javax.sound.midi.MidiDevice.Info;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.dao.DataAccessException;

import org.springframework.data.redis.core.RedisOperations;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.data.redis.core.SessionCallback;

import org.springframework.data.redis.core.script.RedisScript;

public class MultiTestLock implements Lock {

Logger logger=LoggerFactory.getLogger(MultiTestLock.class);

private RedisTemplate<Serializable, Serializable> redisTemplate;

public MultiTestLock(RedisTemplate<Serializable, Serializable> redisTemplate) {

super();

this.redisTemplate = redisTemplate;

}

@Override

public void lock() {

//这里使用while循环强制线程进来之后先进行抢锁操作。只有抢到锁才能进行后续操作

while(true){

if(tryLock()){

try {

//这里让线程睡500毫秒的目的是为了模拟业务耗时,确保业务结束时之前设置的值正好打到超时时间,

//实际生产中可能有偏差,这里需要经验

Thread.sleep(500l);

// logger.info(Thread.currentThread().getName()+" time to awake");

return;

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}else{

try {

//这里设置一个随机毫秒的sleep目的时降低while循环的频率

Thread.sleep(new Random().nextInt(200)+100);

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

}

@Override

public boolean tryLock() {

//这里也可以选用transactionSupport支持事务操作

SessionCallback<Object> sessionCallback=new SessionCallback<Object>() {

@Override

public Object execute(RedisOperations operations)

throws DataAccessException {

operations.multi();

operations.opsForValue().setIfAbsent("secret", "answer");

//设置超时时间要根据业务实际的可能处理时间来,是一个经验值

operations.expire("secret", 500l, TimeUnit.MILLISECONDS);

Object object=operations.exec();

return object;

}

};

//执行两部操作,这里会拿到一个数组值 [true,true],分别对应上述两部操作的结果,如果中途出现第一次为false则表明第一步set值出错

List<Boolean> result=(List) redisTemplate.execute(sessionCallback);

// logger.info(Thread.currentThread().getName()+" try lock "+ result);

if(true==result.get(0)||"true".equals(result.get(0)+"")){

logger.info(Thread.currentThread().getName()+" try lock success");

return true;

}else{

return false;

}

}

@Override

public boolean tryLock(long arg0, TimeUnit arg1)

throws InterruptedException {

// TODO Auto-generated method stub

return false;

}

@Override

public void unlock() {

//unlock操作直接删除锁,如果执行完还没有达到超时时间则直接删除,让后续的线程进行继续操作。起到补刀的作用,确保锁已经超时或被删除

SessionCallback<Object> sessionCallback=new SessionCallback<Object>() {

@Override

public Object execute(RedisOperations operations)

throws DataAccessException {

operations.multi();

operations.delete("secret");

Object object=operations.exec();

return object;

}

};

Object result=redisTemplate.execute(sessionCallback);

}

@Override

public void lockInterruptibly() throws InterruptedException {

// TODO Auto-generated method stub

}

@Override

public Condition newCondition() {

// TODO Auto-generated method stub

return null;

}

public RedisTemplate<Serializable, Serializable> getRedisTemplate() {

return redisTemplate;

}

public void setRedisTemplate(

RedisTemplate<Serializable, Serializable> redisTemplate) {

this.redisTemplate = redisTemplate;

}

}

执行结果

可以看到,票数稳步减少,后续没有抢到锁的线程余票为0,无票可抢。

tips:

这其中也出现了一个问题,redis进行多部封装操作时,系统报错:ERR EXEC without MULTI

后经过查阅发现问题出在:

在spring中,多次执行MULTI命令不会报错,因为第一次执行时,会将其内部的一个isInMulti变量设为true,后续每次执行命令是都会检查这个变量,如果为true,则不执行命令。

而多次执行EXEC命令则会报开头说的"ERR EXEC without MULTI"错误。

以上为个人经验,希望能给大家一个参考,也希望大家多多支持。如有错误或未考虑完全的地方,望不吝赐教。

以上是 利用redis实现分布式锁,快速解决高并发时的线程安全问题 的全部内容, 来源链接: utcz.com/a/254450.html

回到顶部