一、分布式锁基础
1.1线程锁与进程锁
首先说一下线程锁与进程锁的相关知识,这样更便于理解分布式锁。
线程锁:给方法、代码块加锁。加锁后同一时刻仅有一个线程执行该方法或该代码块。线程锁仅在同一JVM中有效。因为线程锁的实现靠线程之间共享内存。如synchronized是共享对象头,显示锁Lock是共享某个变量。
进程锁:为了控制同一操作系统多个进程同时访问某个共享资源,因进程具有独立性,各个进程无法访问其他进程的资源,因此无法通过synchronized等线程锁实现进程锁。
分布式锁与进程锁类似,只是跨多个操作系统。即多个进程不在同一系统中,用分布式锁控制多个进程对资源的访问。其实分布式锁可以实现线程锁和进程锁,但是不建议,因为分布式锁很耗费资源。简单来说,单机情况下即一个JVM,线程之间共享内存,可以通过线程锁解决并发问题。如果是分布式情况下,即多个JVM,线程A和线程B可能不在同一个JVM中,线程锁不起作用 ,只能用分布式锁解决问题。
1.2分布式锁的满足条件
1)系统是分布式系统
2)共享资源,即各个系统访问同一资源
3)同步访问,即很多进程同时访问同一资源
例如,多台tomcat服务器高并发访问同一redis或mysql数据库应该考虑分布式锁。
1.3实现分布式锁的原理
实现分布式锁有多种方式,如zookeeper、redis等,无论哪种方式,基本原理都是:用一个状态值表示锁,对锁的占用和释放通过状态值来改变。通过zookeeper或者redis服务器存储锁信息。本篇讲利用redis实现分布式锁。redis为单进程单线程模式,采用队列模式将并发访问变成串行访问。
二、分布式锁实现方式一
2.1相关基础知识
redis实现分布式锁的两个主要命令如下:
1)setnx(key,value):若key不存在,设置key-value返回1,若key存在,返回0,不做任何操作。
2)getset(key,value):若key不存在,返回nil,设置新的key-value,若key存在,返回旧值,并且设置一个新值。
2.2实现分布式锁思路
1)用一个字符串常量表示锁,即key,用当前时间+锁超时时间当做value。只有一种情况是占有锁的即key在redis中,并且它对应的value大于当前时间。通过不断的设置key或value或删除key来控制线程是否占有锁。
2)首先对key执行setnx操作,如果返回1则获得了锁。否则执行get操作,拿到value,如果value小于当前时间说明锁已过期(可能某个客户端持有锁,但是崩溃了引起的)。接着执行getset操作,这时得到值value2,如果value等于value2即可获得锁(俩值相等说明在这期间没有别的线程执行getset操作,即没有走过这段获得锁的代码。当然后进来的线程可能改变value的值,但误差极小,可忽略不计,后面代码中很明了)
3)循环执行2中的操作,并设置Thread.sleep(100),避免线程饥饿。
4)在finally中要有释放锁的操作。
2.3实现代码
RedisLock.java
public class RedisLock { private static Logger logger = LoggerFactory.getLogger(RedisLock.class); private String lockKey; private int expireMsecs = 60 * 1000; private int timeoutMsecs = 10 * 1000; private volatile boolean locked = false; private RedisTemplate<Object, Object> redisTemplate; public RedisLock(RedisTemplate<Object,Object> redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) { this.redisTemplate = redisTemplate; this.lockKey = lockKey; this.timeoutMsecs = timeoutMsecs; this.expireMsecs = expireMsecs; } public String get(final String key) { Object obj = null; try { obj = redisTemplate.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { StringRedisSerializer serializer = new StringRedisSerializer(); byte[] data = connection.get(serializer.serialize(key)); connection.close(); if (data == null) { return null; } return serializer.deserialize(data); } }); } catch (Exception e) { logger.error("get redis error, key : {}", key); } return obj != null ? obj.toString() : null; } public String set(final String key,final String value) { Object obj = null; try { obj = redisTemplate.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { StringRedisSerializer serializer = new StringRedisSerializer(); connection.set(serializer.serialize(key), serializer.serialize(value)); return serializer; } }); } catch (Exception e) { logger.error("get redis error, key : {}", key); } return obj != null ? obj.toString() : null; } public boolean setNX(final String key, final String value) { Object obj = null; try { obj = redisTemplate.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { StringRedisSerializer serializer = new StringRedisSerializer(); Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value)); connection.close(); return success; } }); } catch (Exception e) { logger.error("setNX redis error, key : {}", key); } return obj != null ? (Boolean) obj : false; } private String getSet(final String key, final String value) { Object obj = null; try { obj = redisTemplate.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { StringRedisSerializer serializer = new StringRedisSerializer(); byte[] ret = connection.getSet(serializer.serialize(key), serializer.serialize(value)); connection.close(); return serializer.deserialize(ret); } }); } catch (Exception e) { logger.error("setNX redis error, key : {}", key); } return obj != null ? (String) obj : null; } public synchronized boolean lock() throws InterruptedException { int timeout = timeoutMsecs; while (timeout >= 0) { long expires = System.currentTimeMillis() + expireMsecs + 1; String expiresStr = String.valueOf(expires); //锁到期时间 if (this.setNX(lockKey, expiresStr)) { locked = true; return true; } String currentValueStr = this.get(lockKey); //redis里的时间 if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) { //超时了 锁失效 String oldValueStr = this.getSet(lockKey, expiresStr); if (oldValueStr != null && oldValueStr.equals(currentValueStr)) { //同一时间只有一个线程走到这里 获得锁 locked = true; return true; } } timeout -= 100; Thread.sleep(100); } return false; } public synchronized void unlock() { if (locked) { redisTemplate.delete(lockKey); locked = false; } } }
BuyService.java
public class BuyService extends Thread{ private RedisTemplate<Object,Object> redisTemplate; private String key; public BuyService(RedisTemplate<Object,Object> redisTemplate,String key) { this.redisTemplate = redisTemplate; this.key = key; } public boolean buy() { RedisLock lock = new RedisLock(redisTemplate, key, 10000, 20000); try { if (lock.lock()) { String goodsNum=lock.get("goodsNum"); if(Integer.parseInt(goodsNum)-1>=0) { lock.set("goodsNum",String.valueOf(Integer.parseInt(goodsNum)-1));//每次减1 System.out.println("线程"+Thread.currentThread().getName()+" 抢购成功!剩余数量:"+(Integer.parseInt(goodsNum)-1)); }else { System.out.println("线程"+Thread.currentThread().getName()+" 没有抢到!"); } return true; } } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } return false; } @Override public void run() { buy(); } }
TestController.java
@RestController public class TestController { @Autowired private RedisTemplate<Object,Object> redisTemplate; @GetMapping("/testlock") public void testSetUser() { for (int i = 0; i < 200; i++) { BuyService buyService = new BuyService(redisTemplate, "MSKEY"); buyService.start(); } } }
RedisConfig.java
@Configuration public class RedisConfig { @Bean @ConditionalOnMissingBean(name = "redisTemplate") public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) { RedisTemplate<Object, Object> template = new RedisTemplate<>(); Jackson2JsonRedisSerializer<Object> fastJsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class); template.setValueSerializer(fastJsonRedisSerializer); template.setHashValueSerializer(fastJsonRedisSerializer); template.setKeySerializer(new StringRedisSerializer()); template.setHashKeySerializer(new StringRedisSerializer()); template.setConnectionFactory(redisConnectionFactory); return template; } }