美文网首页右耳菌-邓小白的Java架构师的修炼之路
Java并发 - 并发工具类

Java并发 - 并发工具类

作者: 右耳菌 | 来源:发表于2022-06-22 15:02 被阅读0次

AQS

可以先查看 Java并发 - 读写锁与AQS简单了解 进行简单的了解

AQS中的队列

通过维护 state 进行加锁和解锁（含读锁（共享锁）和写锁（独占锁））

    /**
     * The synchronization state.
     */
    private volatile int state;

继承 AbstractQueuedSynchronizer 后需要实现的方法

  • isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
  • tryAcquire(int):独占方式。尝试获取资源，成功则返回true，失败则返回false。
  • tryRelease(int):独占方式。尝试释放资源，成功则返回true，失败则返回false。
  • tyAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0表示成功，但没有剩余可用资源;正数表示成功，且有剩余资源。
  • tryReleaseShared(int):共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。

1. Semaphore （依赖了 AQS 定义的模板方法）

Semaphore是一个计数信号量，常用于限制可以访问某些资源（物理或逻辑的）线程数目。
简单说，是一种用来控制并发量的共享锁。


例子 
package concurrent;

import java.util.concurrent.Semaphore;

public class Semaphore_Demo {
     static Semaphore sp =new Semaphore(6);


    public static void main(String[] args) {
        for (int i = 0; i <1000 ; i++) {
            new Thread(()->{
                try {
                    sp.acquire(); //就是拿信号量
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("当前信号量限制....");
                try {
                    Thread.sleep(2000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("执行完了");
                sp.release(); //执行完了，就释放信号量
            }).start();
        }
    }
}

当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
执行完了
执行完了
执行完了
执行完了
当前信号量限制....
执行完了
执行完了
当前信号量限制....
.....

以上例子表明，Semaphore(6) 限制了只能同时有六个线程执行，也就起到了限流的作用

  • 自己实现Semaphore的例子
package concurrent;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;

public class MySemaphore {

    private Sync sync;

    public MySemaphore(int permits) {
        sync = new Sync(permits);
    }

    public void acquire() {
        sync.acquireShared(1);
    }

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


    class Sync extends AbstractQueuedSynchronizer {
        // 信号量大小
        private int permits;

        public Sync(int permits) {
            this.permits = permits;
        }

        @Override
        protected int tryAcquireShared(int arg) {
            //定义自己的方法
            int state = getState();
            int nextState = state + arg;

            if (nextState <= permits) {
                if (compareAndSetState(state, nextState))
                    return 1;
            }
            return -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            int state = getState();
            if (compareAndSetState(state, state - arg)) {
                return true;
            } else {
                return false;
            }

        }
    }
}

执行代码

package concurrent;

public class Semaphore_Demo {
     static MySemaphore sp =new MySemaphore(6);


    public static void main(String[] args) {
        for (int i = 0; i <1000 ; i++) {
            new Thread(()->{
                sp.acquire(); //就是拿信号量
                System.out.println("当前信号量限制....");
                try {
                    Thread.sleep(2000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("执行完了");
                sp.release(); //执行完了，就释放信号量
            }).start();
        }
    }
}

执行结果

当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
当前信号量限制....
执行完了
执行完了
执行完了
执行完了
执行完了
执行完了

2. CountDownLatch (依赖了 AQS 定义的模板方法)

即实现了一种计数器，可以认为是倒数计时，比方说 从 6 开始依次递减（此时只是准备就绪），然后到 0 的时候再一起启动。
好比有这样的一个跑步比赛，比赛开始之前，所有的选手都需要向裁判示意已经准备就绪，一共有六个选手，那么每一个选手向裁判示意准备就绪，未准备就绪的人数就减1，直到所有的选手都准备就绪，此时未准备就绪的人数为0，那么裁判就会扣下发令枪示意比赛开始。

例子 
package concurrent;

import java.util.concurrent.CountDownLatch;

public class CountDownLatch_Demo {

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(6); //计数为6


        for (int i = 0; i <6 ; i++) {
            new Thread(()->{
                System.out.println("开始准备.....");
                latch.countDown();//计数减一
            }).start();
            Thread.sleep(1000);
        }

        latch.await(); //每个线程执行一次，则-1，在latch为0的时候开始向下运行 这是这些线程都准备就绪，然后去一起干同一件事

        //还有一种方式， 将一个活分为多段，每个线程去干一段
//        for (int i = 0; i <6 ; i++) {
//            new Thread(()->{
//                  latch.countDown(); // 计数减一
//                try {
//                    latch.await(); // 阻塞 -- > 0
//                    System.out.println("线程："+Thread.currentThread().getName()+"执行完毕");
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }).start();
//        }

        System.out.println("开始干活....");
    }
}

开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始干活....
  • 自己实现CountDownLatch的例子
package concurrent;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;

public class MyCountDownLatch {

    private Sync sync;

    public MyCountDownLatch(int count) {
        sync = new Sync(count);
    }

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

    public void await() {
        sync.acquireShared(1);
    }


    class Sync extends AbstractQueuedSynchronizer {

        public Sync(int count) {
            setState(count);
        }


        @Override
        protected int tryAcquireShared(int arg) {
            return getState() == 0 ? 1 : -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            while (true) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c - 1;
                if (compareAndSetState(c, nextc)) {
                    return nextc == 0;
                }
            }
        }
    }
}

执行代码

package concurrent;

import java.util.concurrent.CountDownLatch;

public class CountDownLatch_Demo {

    public static void main(String[] args) throws InterruptedException {
        MyCountDownLatch latch = new MyCountDownLatch(6); //计数为6


        for (int i = 0; i <6 ; i++) {
            new Thread(()->{
                System.out.println("开始准备.....");
                latch.countDown();//计数减一
            }).start();
            Thread.sleep(1000);
        }

        latch.await(); //每个线程执行一次，则-1，在latch为0的时候开始向下运行 这是这些线程都准备就绪，然后去一起干同一件事

        //还有一种方式， 将一个活分为多段，每个线程去干一段
//        for (int i = 0; i <6 ; i++) {
//            new Thread(()->{
//                  latch.countDown(); // 计数减一
//                try {
//                    latch.await(); // 阻塞 -- > 0
//                    System.out.println("线程："+Thread.currentThread().getName()+"执行完毕");
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }).start();
//        }

        System.out.println("开始干活....");
    }
}

执行结果

开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始准备.....
开始干活....

3. CyclicBarrier （依赖了可重入锁的Condition和 signalAll）

循环栅栏，可以循环利用的屏障。
举例:排队上摩天轮时，每到齐四个人，就可以上同一个车厢。


例子 
package concurrent;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrier_Demo {

    public static void main(String[] args) throws InterruptedException {
        CyclicBarrier barrier = new CyclicBarrier(4);

        for (int i = 0; i < 100; i++) { //假设有100个任务，每次只能有固定数量的线程去执行，可以使用这个
            new Thread(() -> {
                try {
                    barrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
                System.out.println("任务开始执行");

            }).start();
            Thread.sleep(500L);
        }
    }
}
  • 自己实现CyclicBarrier 的例子
package concurrent;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class MyCyclicBarrier {

    private final ReentrantLock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();

    private int count = 0; //批次

    private final int parties; //多少线程准备就绪？

    private Object generation = new Object(); // 类似于版本号，解决伪唤醒的问题

    public MyCyclicBarrier(int parties) {
        this.parties = parties;
    }


    public void await() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Object g = generation;

            int index = ++count;
            if (index == parties) {
                nextGeneration();
                return;
            }

            while (true) {
                try {
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                if (g != generation) {
                    return;
                }
            }
        } finally {
            lock.unlock();
        }
    }

    public void nextGeneration() {
        condition.signalAll();
        count = 0;
        generation = new Object();
    }
}

执行代码

package concurrent;

public class CyclicBarrier_Demo {

    public static void main(String[] args) throws InterruptedException {
        MyCyclicBarrier barrier = new MyCyclicBarrier(4);

        for (int i = 0; i < 100; i++) { //假设有100个任务，每次只能有固定数量的线程去执行，可以使用这个
            new Thread(() -> {
                barrier.await();
                System.out.println("任务开始执行");
            }).start();
            Thread.sleep(500L);
        }
    }
}

如果觉得有收获就点个赞吧，更多知识，请点击关注查看我的主页信息哦~

相关文章

网友评论

    本文标题：Java并发 - 并发工具类

    本文链接：https://www.haomeiwen.com/subject/mxeyvrtx.html

    延伸阅读

    深度阅读

    • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

    栏目导航

    热点阅读

    右耳菌-邓小白的Java架构师的修炼之路

    关于我们|服务条款|联系我们|Java并发 - 并发工具类|投稿指南|网站地图|RSS订阅|排版工具|手机版

    提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

    Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

    备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

    本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

    所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！