本文主要是介绍java并发编程(7) 共享模型之工具 - stampedLock、semaphore、CountdownLatch、CyclicBarri，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

文章目录

• 前言
• 1. stampedLock
• • 1. 概述
  • 2. 代码
  • • 1. 读读
    • 2. 读写
    • 3. 注意
• 2. Semaphore
• • 1. 基本使用
  • 2. 应用场景
  • 3. 原理
• 3. CountdownLatch
• • 1. 介绍
  • 2. 使用
  • 3. 应用之等待多线程加载完毕
  • 4. 应用之等待多个远程调用结束
• 4. CyclicBarrier
• • 1. 概念
  • 2. 基本使
  • 3. 业务场景

前言

这篇文章讨论读写锁stampedLock。文章根据《Java并发编程的艺术》这本书以及黑马的视频 黑马多线程 做的笔记。

1. stampedLock

1. 概述

该类自 JDK 8 加入，是为了进一步优化读性能，它的特点是在使用读锁、写锁时都必须配合【戳】使用 加解读锁

long stamp = lock.readLock();
lock.unlockRead(stamp);

加解写锁

long stamp = lock.writeLock();
lock.unlockWrite(stamp);

乐观读，StampedLock 支持 tryOptimisticRead() 方法（乐观读），读取完毕后需要做一次 戳校验 如果校验通 过，表示这期间确实没有写操作，数据可以安全使用，如果校验没通过，需要重新获取读锁升级成真正的读锁，保证数据安全。

long stamp = lock.tryOptimisticRead();
// 验戳
if(!lock.validate(stamp)){
 // 锁升级
}

2. 代码

提供一个 数据容器类 内部分别使用读锁保护数据的 read() 方法，写锁保护数据的 write() 方法

class DataContainerStamped {
	//数据
    private int data;
    //StampedLock 锁
    private final StampedLock lock = new StampedLock();

    public DataContainerStamped(int data) {
        this.data = data;
    }
	//读取操作
    public int read(int readTime) {
        //首先获取stamp
        long stamp = lock.tryOptimisticRead();
        log.debug("optimistic read locking...{}", stamp);
        sleep(readTime);
        //验证如果是有效的，证明这期间没有写操作，直接返回即可，这时还是乐观锁
        if (lock.validate(stamp)) {
            //就可以读到数据
            log.debug("read finish...{}, data:{}", stamp, data);
            return data;
        }
        // 否则证明已经有写锁修改过了，这里需要再次获取读锁，升级为真正的读锁
        // 锁升级 - 读锁
        log.debug("updating to read lock... {}", stamp);
        try {
            //获取stamp
            stamp = lock.readLock();
            log.debug("read lock {}", stamp);
            sleep(readTime);
            log.debug("read finish...{}, data:{}", stamp, data);
            return data;
        } finally {
            log.debug("read unlock {}", stamp);
            lock.unlockRead(stamp);
        }
    }

    public void write(int newData) {
        //获取戳
        long stamp = lock.writeLock();
        log.debug("write lock {}", stamp);
        try {
            sleep(2);
            this.data = newData;
        } finally {
            log.debug("write unlock {}", stamp);
            lock.unlockWrite(stamp);
        }
    }
}

1. 读读

public class TestStampedLock {
    public static void main(String[] args) {
        DataContainerStamped dataContainer = new DataContainerStamped(1);
        new Thread(() -> {
            dataContainer.read(1);
        }, "t1").start();
        sleep(0.5);
        new Thread(() -> {
            dataContainer.read(0);
        }, "t2").start();
    }
}

输出结果：下面结果中可以看到两个线程同时获取读锁并执行读操作，没有先后的关系。

19:09:07.857 [t1] DEBUG c.DataContainerStamped - optimistic read locking...256
19:09:08.361 [t2] DEBUG c.DataContainerStamped - optimistic read locking...256
19:09:08.362 [t2] DEBUG c.DataContainerStamped - read finish...256, data:1
19:09:08.873 [t1] DEBUG c.DataContainerStamped - read finish...256, data:1

2. 读写

public class TestStampedLock {
    public static void main(String[] args) {
        DataContainerStamped dataContainer = new DataContainerStamped(1);
        new Thread(() -> {
            dataContainer.read(1);
        }, "t1").start();
        sleep(0.5);
        new Thread(() -> {
            dataContainer.write(0);
        }, "t2").start();
    }
}

结果输出：一开始是读操作先睡眠一秒，在睡眠之前已经获取了戳了，在 t1 线程睡眠期间 t2 线程获取到了写锁，并将数据修改，而且戳也改成了384.此时 t1 线程醒过来校验发现戳已经被修改了，所以这时候 t1 线程会等待 t2 线程释放写锁之后去获取读锁。完成从乐观读 -> 读锁 的升级。

19:10:49.987 [t1] DEBUG c.DataContainerStamped - optimistic read locking...256
19:10:50.485 [t2] DEBUG c.DataContainerStamped - write lock 384
19:10:50.998 [t1] DEBUG c.DataContainerStamped - updating to read lock... 256
19:10:52.498 [t2] DEBUG c.DataContainerStamped - write unlock 384
19:10:52.498 [t1] DEBUG c.DataContainerStamped - read lock 513
19:10:53.508 [t1] DEBUG c.DataContainerStamped - read finish...513, data:0
19:10:53.508 [t1] DEBUG c.DataContainerStamped - read unlock 513

3. 注意

• StampedLock 不支持条件变量（await、signal 这些没法用）
• StampedLock 不支持可重入

2. Semaphore

1. 基本使用

信号量，用来限制能同时访问共享资源的线程上限。它通过协调各个线程，以保证合理的使用公共资源。

@Slf4j
public class TestSemaphore {
    public static void main(String[] args) {
        // 1. 创建 semaphore 对象
        //这里设置上限为3，表示线程只支持三个，达到了3个线程之后这个变量就为0了
        //第二个参数是表示公平非公平：其他线程来了如果是公平是不可能竞争的
        //如果是非公平是可以和等待队列里面的线程竞争的
        Semaphore semaphore = new Semaphore(3);

        // 2. 10个线程同时运行
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    //3. 获取许可，acquire后semaphore-1变成2
                    semaphore.acquire();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    log.debug("running...");
                    sleep(1);
                    log.debug("end...");
                } finally {
                    semaphore.release();
                }
            }).start();
        }
    }
}

结果输出：很明显的是，首先 0，1，2三个线程先执行，然后其他线程暂停，然后线程0，1，2执行完成之后其他线程接着竞争三个名额。

2. 应用场景

Semaphore 可以用于流量控制，特别是公共资源有限的应用场景，比如数据库连接。加入有一个需求，需要读取几万个文件的数据，这属于 IO 密集型任务，我们可以启动几十个线程去读取文件，但是在读取到本地之后要写入数据库时我们就得控制连接数，否则连接数过多会报错无法获取连接。这时候就可以使用 Semaphore 来控制并发数。

• 使用 Semaphore 在高峰时让请求线程阻塞，等到高峰过去了再释放许可，当然它只适合限制单机线程数量（没有考虑分布式），并且仅仅是限制线程数而不是资源数。该处理多少资源还是得处理多少资源
• 使用 Semaphore 简单实现连接池（一个线程对应一个连接），对比享元模式下的实现（wait，notify），性能和可读性更好。

public class TestPoolSemaphore {
    public static void main(String[] args) {
        Pool pool = new Pool(2);
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                Connection conn = pool.borrow();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                pool.free(conn);
            }).start();
        }
    }
}

class Pool {
    // 1. 连接池大小
    private final int poolSize;
    // 2. 连接对象数组
    private Connection[] connections;
    // 3. 连接状态数组 0 表示空闲， 1 表示繁忙
    private AtomicIntegerArray states;
    private Semaphore semaphore;

    // 4. 构造方法初始化
    public Pool(int poolSize) {
        this.poolSize = poolSize;
        // 让许可数与资源数一致，保证一个线程一个资源
        this.semaphore = new Semaphore(poolSize);
        this.connections = new Connection[poolSize];
        this.states = new AtomicIntegerArray(new int[poolSize]);
        for (int i = 0; i < poolSize; i++) {
            connections[i] = new MockConnection("连接" + (i+1));
        }
    }

    // 5. 借连接
    public Connection borrow() {// t1, t2, t3
        // 获取许可
        try {
            semaphore.acquire(); // 没有许可的线程，在此等待
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        for (int i = 0; i < poolSize; i++) {
            // 获取空闲连接
            if(states.get(i) == 0) {
                if (states.compareAndSet(i, 0, 1)) {
                    log.debug("borrow {}", connections[i]);
                    return connections[i];
                }
            }
        }
        // 不会执行到这里，肯定能找到空闲连接的，线程数和连接数一样
        return null;
    }
    // 6. 归还连接
    public void free(Connection conn) {
        for (int i = 0; i < poolSize; i++) {
            if (connections[i] == conn) {
                states.set(i, 0);
                log.debug("free {}", conn);
                //归还许可
                semaphore.release();
                break;
            }
        }
    }
}

class MockConnection implements Connection{}

输出结果：可以看到，这里我们设置了大小为2之后，每次获取都是两个两个获取的。其他线程就在等着，下面就以这个为例，讲讲原理

3. 原理

Semaphore 有点像一个停车场，permits 就好像停车位数量，当线程获得了 permits 就像是获得了停车位，然后停车场显示空余车位减一

1、刚开始，假设permits（state）为 3，这时 5 个线程来获取资源，下面时构造方法的调用链




其实到这里，本质上就是赋值给了 AQS 中的 state

2、假设其中 Thread-1，Thread-2，Thread-4 cas 竞争成功，而 Thread-0 和 Thread-3 竞争失败，这时候就进入 AQS 队列park 阻塞




3、这时 Thread-4 释放了 permits，状态如下

接下来 Thread-0 竞争成功，permits 从1再次设置为 0，设置自己为 head 节点，断开原来的 head 节点，unpark 接下来的 Thread-3 节点，但由于 permits 是 0，因此 Thread-3 在尝试不成功后再次进入 park 状态

加锁

public void acquire() throws InterruptedException {
   sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
      throws InterruptedException {
     if (Thread.interrupted())
         throw new InterruptedException();
     //tryAcquireShared返回值时剩余的资源数，当满了之后就返回负数
     if (tryAcquireShared(arg) < 0)
         doAcquireSharedInterruptibly(arg);
 }

//tryAcquireShared里面调用了一个这个方法
 final int nonfairTryAcquireShared(int acquires) {
     for (;;) {
          //获取状态，还剩多少名额
          int available = getState();
          //减去许可数 3-1 = 2，如果这时候已经为0了，那么就返回负数
          int remaining = available - acquires;
          //设置剩余的许可数
          if (remaining < 0 ||
              compareAndSetState(available, remaining))
              //返回剩余的数
              return remaining;
      }
  }

//doAcquireSharedInterruptibly已经说过了
private void doAcquireSharedInterruptibly(int arg)
  throws InterruptedException {
     //创建 Share 节点
     final Node node = addWaiter(Node.SHARED);
     boolean failed = true;
     try {
         for (;;) {
             //找到前驱节点
             final Node p = node.predecessor();
             //判断是不是头
             if (p == head) {
                 //此时当前线程时老二，尝试再次获取
                 int r = tryAcquireShared(arg);
                 if (r >= 0) {
                     //获取成功了就设置头节点为空，并且唤醒后面所有的共享节点
                     setHeadAndPropagate(node, r);
                     p.next = null; // help GC
                     failed = false;
                     return;
                 }
             }
             //然后在这里park住
             if (shouldParkAfterFailedAcquire(p, node) &&
                 parkAndCheckInterrupt())
                 throw new InterruptedException();
         }
     } finally {
         if (failed)
             cancelAcquire(node);
     }
 }

解锁

public void release() {
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

//释放
 protected final boolean tryReleaseShared(int releases) {
     
     for (;;) {
        //拿到状态，就是剩余的资源数，此时是0
         int current = getState();
         //释放了一个当然要加上1了
         int next = current + releases;
         //如果next < 当前的，证明加法溢出了
         if (next < current) // overflow
             throw new Error("Maximum permit count exceeded");
          //CAS设置
         if (compareAndSetState(current, next))
             return true;
     }
 }

  private void doReleaseShared() {
      //for循环
      for (;;) {
         //获取头节点，如果是tail，证明此时没有其他等待节点了
         Node h = head;
         if (h != null && h != tail) {
             //获取状态
             int ws = h.waitStatus;
             //如果是-1，证明有义务唤醒下一个节点
             if (ws == Node.SIGNAL) {
                 //把状态从 -1 改成 0
                 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                     continue;   
                  //唤醒下一个节点
                 unparkSuccessor(h);
             }
             else if (ws == 0 &&
                      !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                 continue;                // loop on failed CAS
         }
         if (h == head)                   // loop if head changed
             break;
     }
 }

3. CountdownLatch

1. 介绍

用来进行线程同步协作，等待所有线程完成倒计时

其中构造参数用来初始化等待计数值，await() 用来等待计数归零，countDown() 用来让计数 -1

下面是内部的锁的实现

 private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
		//设置状态，就是计数值
        Sync(int count) {
            setState(count);
        }
        
        int getCount() {
            return getState();
        }
        //这个方法就是看看线程有没有都执行完了，如果是0就表示线程都执行完了
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
		//释放锁
        protected boolean tryReleaseShared(int releases) {
            //一个线程用完就让计数值-1
            for (;;) {
                //获取状态
                int c = getState();
                if (c == 0)
                    return false;
                //-1
                int nextc = c-1;
                //CAS修改
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

2. 使用

@Slf4j(topic = "c.TestCountDownLatch")
public class TestCountDownLatch {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        CountDownLatch latch = new CountDownLatch(3);
        new Thread(()->{
            log.debug("线程 t1 开始");
            sleep(1);
            latch.countDown();
            log.debug("线程 t1 结束");
        }, "t1").start();
        new Thread(()->{
            log.debug("线程 t2 开始");
            sleep(2);
            latch.countDown();
            log.debug("线程 t2 结束");
        }, "t2").start();
        new Thread(()->{
            log.debug("线程 t3 开始");
            sleep(5);
            latch.countDown();
            log.debug("线程 t3 结束");
        }, "t3").start();

        log.debug("主线程等待其他线程执行完成再往下执行");
        latch.await();
        log.debug("主线程执行完成");
    }

输出结果：可以看到下面主线程是等待其他线程执行完成了才继续往下执行的，至于下面的 t3 结束打印在最后是因为 t3 结束先唤醒了主线程

其实后来的线程的使用基本都是使用线程池的，所以线程一般不会轻易结束，这时候用 join 那些 api 就不行了。

我们来看下面的线程池用法：我们使用三个线程来执行任务，然后一个线程等待任务结束返回结果

@Slf4j(topic = "c.TestCountDownLatch")
public class TestCountDownLatch {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        CountDownLatch latch = new CountDownLatch(3);
        ExecutorService service = Executors.newFixedThreadPool(4);
        service.submit(()->{
            log.debug("线程开始");
            sleep(1);
            latch.countDown();
            log.debug("线程结束");
        });
        service.submit(()->{
            log.debug("线程开始");
            sleep(1.5);
            latch.countDown();
            log.debug("线程结束");
        });
        service.submit(()->{
            log.debug("线程开始");
            sleep(2);
            latch.countDown();
            log.debug("线程结束");
        });
        service.submit(()->{
            try {
                log.debug("线程等待结果");
                latch.await();
                log.debug("线程获取到结果");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
}

输出结果：

3. 应用之等待多线程加载完毕

我们定义下面这个方法，然后实现10个用户加载的情况，等到加载完成打印一句游戏开始

 private static void wangzherongyao() throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(10);
        CountDownLatch latch = new CountDownLatch(10);
        Random r = new Random();
        String[] all = new String[10];
        //10个玩家
        for (int j = 0; j < 10; j++) {
            //final是因为lambda表达式接收的是常量
            final int k = j;
            service.submit(()->{
                for (int i = 0; i <= 100; i++) {
                    try {
                        Thread.sleep(r.nextInt(100));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    all[k] = i + "%";
                    //使用 \r 可以回退到最开始的输出位置然后对原来的输出进行覆盖
                    System.out.print("\r" + Arrays.toString(all));
                }
                //一个任务运行结束了就 -1
                latch.countDown();
            });
        }
        //主线程等待任务执行完成
        latch.await();
        System.out.println("\n游戏开始");
    }

最终输出结果：

4. 应用之等待多个远程调用结束

下面是使用restTemplate来发送请求，在工作中的应用可以是当要不断一个接口请求多此的时候，此时可以用到多线程配合CountDown 来进行计数，当读取完成接口之后就可以对这些获取到的数据进行调用

private static void waitForObject() throws Exception {
        RestTemplate restTemplate = new RestTemplate();
        log.debug("等待远程服务调用返回结果");
        ExecutorService service = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(4);
        service.submit(() -> {
            Map<String, Object> response = restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
            latch.countDown();
        });
        service.submit(() -> {
            Map<String, Object> response1 = restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
            latch.countDown();
        });
        service.submit(() -> {
            Map<String, Object> response1 = restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
            latch.countDown();
        });
        service.submit(() -> {
            Map<String, Object> response3 = restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
            latch.countDown();
        });

        latch.await();
        log.debug("执行完毕");
        service.shutdown();
    }

如果想要返回结果，那么还是使用 Future 更合使：

4. CyclicBarrier

1. 概念

循环栅栏，用来进行线程协作，等待线程满足某个计数。构造时设置『计数个数』，每个线程执行到某个需要“同步”的时刻调用 await() 方法进行等待，当等待的线程数满足『计数个数』时，继续执行。

为什么要用 CyclicBarrier，这个类作用和 CountdownLatch 类似，但是不同点就在于 CyclicBarrier是支持重用的，使用里面的 reset 方法进行重置。所以 CyclicBarrier可以处理更加复杂的业务，比如在计数错误的时候可以进行重置。此外 CyclicBarrier提供了 getNumberWaiting 方法可以获取阻塞的线程个数。isBroken() 方法可以用来了解阻塞的线程是否被中断

2. 基本使

private static void test2() {
        //创建大小为3的线程池对象
        ExecutorService service = Executors.newFixedThreadPool(2);
        //我们设置计数为2，第二个参数是任务，实际是其他两个任务执行完成之后会执行这个任务
        CyclicBarrier barrier = new CyclicBarrier(2, ()-> {
            log.debug("任务1任务2结束");
        });
        for (int i = 0; i < 3; i++) { // task1  task2  task1
            service.submit(() -> {
                log.debug("任务1开始运行");
                sleep(1);
                try {
                    //调用 await 方法进行等待
                    barrier.await(); // 2-1=1
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
            service.submit(() -> {
                log.debug("任务2开始运行");
                sleep(2);
                try {
                    barrier.await(); // 1-1=0
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
        service.shutdown();
    }

输出结果：

注意线程池的数量和计数值要一样，上面这个例子，如果设置线程为3，那么一次执行的顺序是 第1个任务1、第1个任务2 和 第2个任务1，注意任务二这时睡眠了2s，也就是说三个线程执行的时间刚刚好是2s，和线程数是2的时候的结果是一样的，此时执行完成后计数器重置。这样就不对了，这样就是第一个任务2 和 第二个 任务1 让计数器变为 0.

3. 业务场景

使用《Java并发编程的艺术》这本书里面给出的例子。这个计数器可以用于多线程计算数据，最终合并计算结果的场景。例如，用一个 Excel 保存了所有的银行流水，每个 Sheet 保存一个账户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个 sheet 里的银行流水，都执行完成之后，得到每个 sheet 的日军银行流水，最后，再用 barrierAction 用这些线程的计算结果，计算出整个 Excel 的日均 银行流水，代码如下：

public class BankWaterService implements Runnable{
    /**
     * 创建4个屏障，处理完之后执行当前类的 run 方法
     */
    private CyclicBarrier c = new CyclicBarrier(4, this);

    /**
     * 假设只有四个 sheet，那么启动四个线程
     */
    private Executor executor = Executors.newFixedThreadPool(4);

    /**
     * 保存每个sheet处理出来的结果
     */
    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<>();

    private void count(){
        for (int i = 0; i < 4; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    //计算当前sheet的影流数据，代码略,下面直接模拟结果
                    sheetBankWaterCount.put(Thread.currentThread().getName(), 1);
                    try {
                        c.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    @Override
    public void run() {
        int result = 0;
        //汇总每个sheet计算出的结果
        for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
            result += sheet.getValue();
        }
        //将结果输出
        sheetBankWaterCount.put("result", result);
        System.out.println(result);
    }

    public static void main(String[] args) {
        BankWaterService bankWaterService = new BankWaterService();
        bankWaterService.count();   //4
    }
}

如有错误，欢迎指出！！！

这篇关于java并发编程(7) 共享模型之工具 - stampedLock、semaphore、CountdownLatch、CyclicBarri的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！