一 问题背景
我们做的是医疗信息化系统，在系统中一条患者信息对医院中当前科室中的所有诊断医生是可见的，当有一个诊断医生点击按钮处理该数据时，数据的状态发生了变化，其他的医生就不可以再处理此患者的数据了。我们开始的做法是，在医生点击按钮时先去后台数据库获取当前数据状态，根据状态判断数据是否可以操作，如果可以操作，则修改数据状态，进行业务逻辑处理，否则提示数据已被其他人处理，不能处理。
二 问题分析
按照上边的业务逻辑，我们画个图分析，如下图
在上图中，如果用户A和B同时向数据库发起请求获取数据状态，数据库返回wait，A和B都拿到了相同的状态，判断是可以操作数据的，这时他们处理数据。A用户处理完成后提交了数据，数据库状态变为done，记录此数据的处理人为A。由于B用户也可以处理数据，所以他也提交数据，这时数据的操作人记录为了B。有人会说，在A和B提交数据修改状态时再做一个状态的判断，这种也难以避免最开始的获取状态的问题，即使这一步状态获取到了，提示后边的人不能修改，这又会产生系统不友好的问题(我操作了半天，到最后你告诉我不能处理，我白忙活了)。以上问题产生的主要原因就是在多线程情况下对共享数据的资源竞争处理不当，我们需要保证数据的唯一性，即在某一时刻，只能有一个线程独享数据资源。
三 问题解决
如何解决呢？分布式锁，分布式锁有多种实现方式，本文我们用redis实现。由于redis是单线程的，所以一次只能处理一个请求，并将资源分配给这个请求，我们称加 锁。如下图
多线程情况下，redis只会处理其中的一个，其他的暂时等待。如上图当A和B同时发出请求时，redis接受并处理A请求，此时B请求排队等待，等到A请求处理完后再处理B请求。此时redis已经将资源(lock)分配给了A，A请求数据库，B请求没有获取到资源直接返回不在请求数据库。这样就保证了数据库共享数据被唯一资源使用。代码简单实现
1 public class RedisLock { 2 3 private static final String GET_RESULT = "OK"; 4 private static final String RELEASE_RESULT = "1"; 5 private static final String SET_IF_NOT_EXIST = "NX"; 6 private static final String SET_WITH_EXPIRE_TIME = "PX"; 7 8 /** 9 * 获取redis锁 10 * @param jedis redis客户端 11 * @param lockKey 锁标识 key 12 * @param requestId 锁的持有者,加锁的请求 13 * @param expireTime 锁过期时间 14 * @return 15 */ 16 public static boolean getLock(Jedis jedis, String lockKey, String requestId, int expireTime){ 17 //SET_IF_NOT_EXIST 当key不存在时 才处理 18 //SET_WITH_EXPIRE_TIME 设置过期时间 时间由expireTime决定 19 String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime); 20 if (GET_RESULT.equals(result)) { 21 return true; 22 } 23 return false; 24 } 25 26 /** 27 * 释放锁 28 * @param jedis 29 * @param lockKey 30 * @param requestId 31 * @return 32 */ 33 public static boolean releaseLock(Jedis jedis, String lockKey, String requestId){ 34 // 方式1 35 // if (jedis.get(lockKey).equals(requestId)) {//校验当前锁的持有人与但概念请求是否相同 36 // 执行在这里时，如果锁被其它请求重新获取到了，此时就不该删除了 37 // jedis.del(lockKey); 38 // } 39 40 //方式2 41 // eval() 方法会交给redis服务端执行，减少了从服务端再到客户端处理的过程 42 //赋值 KEYS[1] = lockKey ARGV[1] = requestId 43 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; 44 Object releaseResult = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId)); 45 if (RELEASE_RESULT.equals(releaseResult.toString())) { 46 return true; 47 } 48 return false; 49 } 50 }
四 测试锁机制
测试并发我们可以使用一些软件，比如Jmeter，本文我们写个方法测试
1 public static void main(String[] args) { 2 //要创建的线程的数量 3 CountDownLatch looker = new CountDownLatch(1); 4 CountDownLatch latch = new CountDownLatch(10); 5 final String key = "lockKey"; 6 for(int i=0; i < latch.getCount(); i++){ 7 Jedis jedis = new Jedis(); 8 UUID uuid = UUID.randomUUID(); 9 Thread thread = new Thread(new Runnable() { 10 @Override 11 public void run() { 12 try { 13 looker.await(); 14 System.out.println(Thread.currentThread().getName()+"竞争资源，获取锁"); 15 boolean getResult = getLock(jedis, key, uuid.toString(), 5000); 16 if(getResult){ 17 System.out.println(Thread.currentThread().getName()+"获取到了锁，处理业务，用时3秒"); 18 Thread.sleep(3000); 19 boolean releaseResult = releaseLock(jedis, key, uuid.toString()); 20 if(releaseResult){ 21 System.out.println(Thread.currentThread().getName()+"业务处理完毕，释放锁"); 22 } 23 }else{ 24 System.out.println(Thread.currentThread().getName()+"竞争资源失败，未获取到锁"); 25 } 26 latch.countDown(); 27 } catch (InterruptedException e) { 28 e.printStackTrace(); 29 } 30 } 31 }); 32 thread.start(); 33 } 34 35 try { 36 System.out.println("准备，5秒后开始"); 37 Thread.sleep(5000); 38 looker.countDown(); //发令 let all threads proceed 39 40 latch.await(); // // wait for all to finish 41 System.out.println("结束"); 42 } catch (InterruptedException e) { 43 e.printStackTrace(); 44 } 45 46 }
可以看到控制台上输出的结果