多线程下锁的应用

Love The Way You Lie 2022-05-30 01:49 151阅读 0赞

--------------------

###### 给女朋友上锁 ######

> 有一天梦见女朋友跟一个陌生男人逛街，我很是着急，于是有很多志同道合的朋友开始为我出谋划策。有说，让那个男的指向null，让垃圾回收他。 也有的说给个死循环，让他们逛到累，累死他们。。。没错，你们说的都有道理，但是，如果换是我，我会给自己女朋友逛街这个行为上锁，并且只有我才能获取到锁，也不会把锁让给别人。好了，扯完，开始进入正题，没错，就是锁。

相信大家在开发过程中都会遇到这样一种场景：浏览器端发起一个请求，服务器接收到请求后要去数据库中把数据加载回来，对数据做处理，是不是很简单？但是当请求过于频繁或请求量比较大，并且数据库表数据量又大的情况，这时候我们就不得不去关心它的响应时间、性能怎么样等等这些问题了，因为它可能会影响到整个业务流程，甚至整个系统。为解决这个问题，相信很多人会想到用多线程来解决。即在程序中开多个线程并发去处理请求，也就是**并发编程**，从用户角度看还是一个串行过程，实际上是并发在处理，很显然这样做可以提升响应效率。但这又会引起另外一个问题，那就是线程安全，并发编程有三要素：**原子性、有序性、可见性**。在多线程编程中要遵守好这三大要素，如果程序没有处理好，可能会造成一些意想不到的后果……

###### 针对这些问题，jdk提供了一些线程安全的接口和类，例如我们熟悉的Vector、HashTable。java还提供了synchronized关键字和修饰符valitate来保证线程安全，这两种都是利用锁来实现的。 ######

*  valitate  
    被valitate修饰的变量，当一个线程改变了它的值时，在内存中相对其他线程来说是可见的。还可以防止重排，即程序按代码的顺序来执行，防止顺序被打乱。但是它并不能保证原子性。
 *  synchronized关键字  
    synchronized是jdk定义实现的锁，确保程序能够对同步块或方法互斥访问，即当一个线程获取到锁后，别的线程只能等待，可以保证原子性。  
    本文主要是对synchronized关键字来展开说明，以下为synchronized的一个例子：

> **案例一**  
> 大量请求同时读取数据库表load回数据写到一个文件（比如excel）中，这个表是分布在不同的库中，并且分表的。若单线程去处理这样的请求，势必会耗时比较久，甚至因为一些慢查询导致连接耗尽，造成严重后果。

思路：  
使用ExecutorService接口来实现多线程，ExecutorService是在包java.util.concurrent下Java中对线程池定义的一个接口，实现异步执行的机制，让任务在后台执行。

ExcelUtils.java

// 使用poi的SXSSFWorkbook支持导出大数据
     private SXSSFWorkbook sxworkBook; 
     private OutputStream out;
    
     public ExcelBuilder(String path, String fileName) {
       sxworkBook = new SXSSFWorkbook(1000);
       try {
          File file = new File(path);
          if (!file.exists()) {
              file.mkdirs();
          }
          File savePath = new File(path + "/" + fileName);
          this.out = new FileOutputStream(savePath);
       } catch (FileNotFoundException e) {
      }
     }
    
    public <T> void writeFile(String sheetname, List<T> dataList, Class<T> clazz) {
       …… 此处写文件，具体参考POI写文件API
    }
    
    public void create(){
     try {
            sxworkBook.write(out);
            out.flush();
        } catch (IOException e) {
             logger.error(e.getMessage(), e);
        } finally {
             // IOUtils.closeQuietly(out);
        }
    }

ThreadTask.java

String dir = "C:/director";
    String fileName = "xxxxx.xlsx";
    ExcelUtils instance = new ExcelUtils(dir, fileName);
    ExecutorService executor = Executors.newCachedThreadPool();
    
    // 开两个线程处理
    executor.execute(new InnerThread(queryParam1, instance));
    executor.execute(new InnerThread(queryParam2, instance));
    
    executor.shutdown();
    while(!executor.awaitTermination(1, TimeUnit.SECONDS));
    
    instance.create();
    
    
    
    
    // 定义一个内部线程类 
    class InnerThread implements Runnable{ 
      private Object queryParam;
      private ExcelUtils instance;
      Object obj = new Object();
    
      InnerThread(Object queryParam, ExcelUtils instance){
        this.queryParam = queryParam;
        this.instance = instance;
      }
    
      @Override
      public void run() {
         List<XXX> lists = service.query(queryParam); // service是查询接口类
         synchronized(obj){
            write(lists);
         }
      }
    
      public void write(List<Vo> lists){
         logger.info("当前线程：" + Thread.currentThread().getName());
         instance.writeFile(System.currentTimeMillis() + "", lists, XXX.class);
      }
    }

到这里就可以实现多线程读DB加载数据，多线程写文件了，但是，事实并不是那样，报异常了！！！！！！  
![1240][]

很明显，write方法使用关键字synchronized加锁无效， 多个线程同时进入同时写文件，从而导致出现异常。那么为什么加了synchronized还出现并发写呢？ 先看看synchronized的定义，如下：（本段摘自百度百科）  
Java语言的关键字，可用来给对象和方法或者代码块加锁，当它锁定一个方法或者一个代码块的时候，同一时刻最多只有一个线程执行这段代码。当两个并发线程访问同一个对象object中的这个加锁同步代码块时，一个时间内只能有一个线程得到执行。另一个线程必须等待当前线程执行完这个代码块以后才能执行该代码块。  
原来synchronized修饰的代码块，必须要实例相同才能锁住代码块，也就是只有一个线程可以执行该块内容。  
于是做了以下修改：

class InnerThread implements Runnable{ 
      private Object queryParam;
      private ExcelUtils instance;
      private Lock lock;
    
      InnerThread(Object queryParam, ExcelUtils instance, Lock lock){
        this.queryParam = queryParam;
        this.instance = instance;
        this.lock = lock; // 传入同一个实例
      }
    
      @Override
      public void run() {
         List<XXX> lists = service.query(queryParam); // service是查询接口类
         synchronized(lock){
            write(lists);
         }
      }
    
      public void write(List<Vo> lists){
         logger.info("当前线程：" + Thread.currentThread().getName());
         instance.writeFile(System.currentTimeMillis() + "", lists, XXX.class);
      }
    }

Lock lock = new ReentrantLock();
    // 开两个线程处理
    executor.execute(new InnerThread(queryParam1, instance, lock));
    executor.execute(new InnerThread(queryParam2, instance, lock));

再次执行，生成文件成功。  
![1240 1][]

注意到一点，这里只对写文件部分加了锁，对于读DB加载数据返回并没有加锁，load数据依然是多线程并行去请求DB，在响应效率上得到了较高提升。

在上面的单机场景中，我们可以运用ava中提供的很多并发处理相关的API，但是这些API在分布式场景中就无能为力了，由于分布式系统的分布性，即多线程和多进程并且分布在不同机器中，synchronized这种锁将失去原有锁的效果，这时候就需要我们自己实现分布式锁。

> **案例二**  
> redis基于缓存，实现分布式锁，利用redis的锁机制来实现分布式锁。

思路：  
利用redis接口API对对象就行上锁，并且设置过期时间，实现并发编程。

RedisClient.java

public class RedisClient{
        private Logger log = LoggerFactory.getLogger(RedisClient.class);
        private JedisPool jedisPool;
        private Jedis jedis;
        String lock;
        long expires = 5000;
    
        public RedisClient(String lock) {
           this.lock = lock;
           this.init();
        }
        private void init() {
            // 池基本配置
            JedisPoolConfig config = new JedisPoolConfig();
            // config.setMaxActive(20);
            config.setMaxIdle(5);
            // config.setMaxWait(1000l);
            config.setTestOnBorrow(false);
    
            jedisPool = new JedisPool(config, "XX.XXX.XXX.XX", 6400);
            jedis = jedisPool.getResource();
        }
    
        public boolean getLock() {
            while (true) {
                boolean lock = setlock();
                if (lock) {
                    return lock;
                }
            }
        }
    
        public boolean setlock() {
            long currentTime = System.currentTimeMillis();
            String expire = String.valueOf(currentTime + expires);
            if (jedis.setnx(lock, expire) > 0) {
                log.info("当前线程：" + Thread.currentThread().getName() + "获取到锁");
                jedis.expire(lock, 5);
                return true;
            } else {
                String oldTime = jedis.get(lock);
                if (oldTime != null && (currentTime - Long.parseLong(oldTime)) > 0) {
                    String oldValue = jedis.getSet(lock, expire);
                    if (oldValue != null && oldValue.equals(oldTime)) {
                        jedis.expire(lock, 5);
                        log.info("过期了，让其它线程获取锁，当前线程：" + Thread.currentThread().getName() + "获取到锁");
                        return true;
                    }
                }
            }
    
            return false;
        }
    }

ThreadTest.java

public class ThreadTest extends Thread { 
        private Logger log = LoggerFactory.getLogger(Main.class);
        String lock;
    
        ThreadTest(String lock) {
            this.lock = lock;
        }
    
        @Override
        public void run() {
            RedisClient client = new RedisClient(lock);
            boolean hasLock = client.getLock();
            if (hasLock) {
                log.info(Thread.currentThread().getName() + "开始执行……");
            }
        }
    }

Main.java

public class Main {
        public static void main(String[] args) {
            String lock = "key";
            new ThreadTest(lock).start();
            new ThreadTest(lock).start();
            new ThreadTest(lock).start();
        }
    
    }

运行效果：  
![1240 2][]

--------------------

总结：  
随着日益增长的业务量，数据越来越大，处理请求响应速度也要随着提升，并发编程就必不可少，在单机多线程下，我们可以使用jdk提供的并发处理相关的API来解决我们的问题，例如上面提到的valitate，synchronized，还有CAS，也可以在业务上实现锁的机制，比如说在数据库变层面来处理。 但在也正是因为业务量越来越大，需求更复杂的前提下，系统分布式部署就越来越重要，负载均衡，多节点，多机器部署势必带来更多的问题，因此分布式锁就广泛被使用，多种实现也随之出现，如前面提到的基于缓存redis，还有基于Zookeeper实现分布式锁等等。

[1240]: /images/20220530/bd81f32296a84fb9b11607158b455e17.png
[1240 1]: /images/20220530/dc64d79a991d42d0b04eaecb259d5960.png
[1240 2]: /images/20220530/479c0313f9634000b2e5cd2cd92a9833.png