利用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"错误。

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