首页 > 其他分享 >通信工具类

通信工具类

时间：2024-07-27 11:07:35浏览次数：10  
标签：Phaser Thread int 通信 线程 new 工具 public

类 作用
Semaphore 限制线程的数量
Exchanger 两个线程交换数据
CountDownLatch 线程等待直到计数器减为 0 时开始工作
CyclicBarrier 作用跟 CountDownLatch 类似，但是可以重复使用
Phaser 增强的 CyclicBarrier

Semaphore

Semaphore 翻译过来是信号的意思。顾名思义，这个工具类提供的功能就是多个线程彼此“传信号”。而这个“信号”是一个int类型的数据，也可以看成是一种“资源”。

// 传入初始资源总数，默认情况下，是非公平的同步器
public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

最主要的方法是 acquire 方法和 release 方法。acquire()方法会申请一个 permit，而 release 方法会释放一个 permit。当然，你也可以申请多个 acquire(int permits)或者释放多个 release(int permits)。

每次 acquire，permits 就会减少一个或者多个。如果减少到了 0，再有其他线程来 acquire，那就要阻塞这个线程直到有其它线程 release permit 为止。

Semaphore 往往用于资源有限的场景中，去限制线程的数量。举个例子，我想限制同时只能有 3 个线程在工作：

public class Main {
    static class MyThread implements Runnable {

        private final int value;
        private final Semaphore semaphore;

        public MyThread(int value, Semaphore semaphore) {
            this.value = value;
            this.semaphore = semaphore;
        }

        @Override
        public void run() {
            try {
                // 获取 permit
                semaphore.acquire();
                System.out.printf("当前线程是%d, 还剩%d个资源，还有%d个线程在等待%n",
                        value, semaphore.availablePermits(), semaphore.getQueueLength());
                // 睡眠随机时间，打乱释放顺序
                Random random = new Random();
                Thread.sleep(random.nextInt(1000));
                System.out.printf("线程%d释放了资源%n", value);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 释放 permit
                semaphore.release();
            }
        }
    }

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < 10; i++) {
            new Thread(new MyThread(i, semaphore)).start();
        }
    }
}

Semaphore 默认的 acquire 方法是会让线程进入等待队列，且抛出异常中断。但它还有一些方法可以忽略中断或不进入阻塞队列：

// 忽略中断
public void acquireUninterruptibly()
public void acquireUninterruptibly(int permits)

// 不进入等待队列，底层使用CAS
public boolean tryAcquire
public boolean tryAcquire(int permits)
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException
public boolean tryAcquire(long timeout, TimeUnit unit)

Semaphore 内部有一个继承了 AQS 的同步器 Sync，重写了tryAcquireShared方法。在这个方法里，会去尝试获取资源。如果获取失败（想要的资源数量小于目前已有的资源数量），就会返回一个负数（代表尝试获取资源失败）。然后当前线程就会进入 AQS 的等待队列。

Exchanger

Exchanger 类用于两个线程交换数据。它支持泛型，也就是说可以在两个线程之间传送任何数据。

public class Main {
    public static void main(String[] args) throws InterruptedException {
        Exchanger<String> exchanger = new Exchanger<>();

        new Thread(() -> {
            try {
                System.out.println("这是线程A，得到了另一个线程的数据："
                        + exchanger.exchange("这是来自线程A的数据"));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 当一个线程调用 exchange 方法后，会处于阻塞状态，只有当另一个线程也调用了 exchange 方法，它才会继续执行
        System.out.println("这个时候线程A是阻塞的，在等待线程B的数据");
        Thread.sleep(1000);

        new Thread(() -> {
            try {
                System.out.println("这是线程B，得到了另一个线程的数据："
                        + exchanger.exchange("这是来自线程B的数据"));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

根据 JDK 里面注释的说法，可以总结为一下特性：

  • 此类提供对外的操作是同步的；
  • 用于成对出现的线程之间交换数据；
  • 可以视作双向的同步队列；
  • 可应用于基因算法、流水线设计等场景。

Exchanger 类还有一个有超时参数的方法，如果在指定时间内没有另一个线程调用 exchange，就会抛出一个超时异常。

public V exchange(V x, long timeout, TimeUnit unit)

Exchanger 是可以重复使用的。也就是说。两个线程可以使用 Exchanger 在内存中不断地再交换数据。

CountDownLatch

public class CountDownLatch {

    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        // 计数值（count）实际上就是闭锁需要等待的线程数量。这个值只能被设置一次，而且 CountDownLatch没有提供任何机制去重新设置这个计数值
        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

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

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

    private final Sync sync;

    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

    // 等待
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    // 超时等待
    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

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

    // 获取当前还有多少count
    public long getCount() {
        return sync.getCount();
    }

    public String toString() {
        return super.toString() + "[Count = " + sync.getCount() + "]";
    }
}

public class Main {
    // 定义前置任务线程
    static class PreTaskThread implements Runnable {
        private final String task;
        private final CountDownLatch countDownLatch;

        public PreTaskThread(String task, CountDownLatch countDownLatch) {
            this.task = task;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                Random random = new Random();
                Thread.sleep(random.nextInt(1000));
                System.out.println(task + " - 任务完成");
                countDownLatch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        // 假设有三个模块需要加载
        CountDownLatch countDownLatch = new CountDownLatch(3);

        // 主任务
        new Thread(() -> {
            try {
                System.out.println("等待数据加载...");
                System.out.printf("还有%d个前置任务%n", countDownLatch.getCount());
                countDownLatch.await();
                System.out.println("数据加载完成，正式开始游戏！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 前置任务
        new Thread(new PreTaskThread("加载地图数据", countDownLatch)).start();
        new Thread(new PreTaskThread("加载人物模型", countDownLatch)).start();
        new Thread(new PreTaskThread("加载背景音乐", countDownLatch)).start();
    }
}

CyclicBarrier

CyclicBarrirer 从名字上来理解是“循环屏障”的意思。前面提到了 CountDownLatch 一旦计数值count被降为 0 后，就不能再重新设置了，它只能起一次“屏障”的作用。而 CyclicBarrier 拥有 CountDownLatch 的所有功能，还可以使用reset()方法重置屏障。

如果参与者（线程）在等待的过程中，Barrier 被破坏，就会抛出 BrokenBarrierException。可以用isBroken()方法检测 Barrier 是否被破坏。

  1. 如果有线程已经处于等待状态，调用 reset 方法会导致已经在等待的线程出现 BrokenBarrierException 异常。并且由于出现了 BrokenBarrierException，将会导致始终无法等待。
  2. 如果在等待的过程中，线程被中断，会抛出 InterruptedException 异常，并且这个异常会传播到其他所有的线程。
  3. 如果在执行屏障操作过程中发生异常，则该异常将传播到当前线程中，其他线程会抛出 BrokenBarrierException，屏障被损坏。
  4. 如果超出指定的等待时间，当前线程会抛出 TimeoutException 异常，其他线程会抛出 BrokenBarrierException 异常。
// 构造方法
public CyclicBarrier(int parties) {
    this(parties, null);
}

public CyclicBarrier(int parties, Runnable barrierAction) {
    // 具体实现
}

public class Main {
    static class PreTaskThread implements Runnable {

        private final String task;
        private final CyclicBarrier cyclicBarrier;

        public PreTaskThread(String task, CyclicBarrier cyclicBarrier) {
            this.task = task;
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run() {
            // 假设总共三个关卡
            for (int i = 1; i < 4; i++) {
                try {
                    Random random = new Random();
                    Thread.sleep(random.nextInt(1000));
                    System.out.printf("关卡%d的任务%s完成%n", i, task);
                    // 一旦调用 await 方法的线程数量等于构造方法中传入的任务总量（这里是 3），就代表达到屏障了。
                    cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        // CyclicBarrier 允许我们在达到屏障的时候可以执行一个任务，可以在构造方法传入一个 Runnable 类型的对象。
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
            System.out.println("本关卡所有前置任务完成，开始游戏...");
        });

        new Thread(new PreTaskThread("加载地图数据", cyclicBarrier)).start();
        new Thread(new PreTaskThread("加载人物模型", cyclicBarrier)).start();
        new Thread(new PreTaskThread("加载背景音乐", cyclicBarrier)).start();
    }
}

CyclicBarrier 内部使用的是 Lock + Condition 实现的等待/通知模式。详情可以查看这个方法的源码：

private int dowait(boolean timed, long nanos)

Phaser

Phaser 是 Java 7 中引入的一个并发同步工具，它提供了对动态数量的线程的同步能力，这与 CyclicBarrier 和 CountDownLatch 不同，因为它们都需要预先知道等待的线程数量。Phaser 是多阶段的，意味着它可以同步不同阶段的多个操作。

前面我们介绍了 CyclicBarrier，可以发现它在构造方法里传入了“任务总量”parties之后，就不能修改这个值了，并且每次调用await()方法也只能消耗一个parties计数。但 Phaser 可以动态地调整任务总量！

Phaser 是阶段性的，所以它有一个内部的阶段计数器。每当我们到达一个阶段的结尾时，Phaser 会自动前进到下一个阶段。

名词解释：

  • Party：Phaser 的上下文中，一个 party 可以是一个线程，也可以是一个任务。当我们在 Phaser 上注册一个 party 时，Phaser 会递增它的参与者数量。
  • arrive：对应一个 party 的状态，初始时是 unarrived，当调用arriveAndAwaitAdvance()或者 arriveAndDeregister()进入 arrive 状态，可以通过getUnarrivedParties()获取当前未到达的数量。
  • register：注册一个新的 party 到 Phaser。
  • deRegister：减少一个 party。
  • phase：阶段，当所有注册的 party 都 arrive 之后，将会调用 Phaser 的onAdvance()方法来判断是否要进入下一阶段。

Phaser 的终止有两种途径，Phaser 维护的线程执行完毕或者onAdvance()返回true。

public class Main {
    static class PreTaskThread implements Runnable {
        private final String task;
        private final Phaser phaser;

        public PreTaskThread(String task, Phaser phaser) {
            this.task = task;
            this.phaser = phaser;
        }

        @Override
        public void run() {
            for (int i = 1; i < 4; i++) {
                try {
                    // 从第二个关卡起，不加载新手教程
                    if (i >= 2 && "加载新手教程".equals(task)) {
                        continue;
                    }
                    Random random = new Random();
                    Thread.sleep(random.nextInt(1000));
                    System.out.printf("关卡%d，需要加载%d个模块，当前模块【%s】%n",
                            i, phaser.getRegisteredParties(), task);

                    // 从第二个关卡起，不加载新手教程
                    if (i == 1 && "加载新手教程".equals(task)) {
                        System.out.println("下次关卡移除加载【新手教程】模块");
                        // 移除一个模块
                        phaser.arriveAndDeregister();
                    } else {
                        phaser.arriveAndAwaitAdvance();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        Phaser phaser = new Phaser(4) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.printf("第%d次关卡准备完成%n", phase + 1);
                return phase == 3 || registeredParties == 0;
            }
        };

        new Thread(new PreTaskThread("加载地图数据", phaser)).start();
        new Thread(new PreTaskThread("加载人物模型", phaser)).start();
        new Thread(new PreTaskThread("加载背景音乐", phaser)).start();
        new Thread(new PreTaskThread("加载新手教程", phaser)).start();
    }
}

这里要注意关卡 1 的输出，在“加载新手教程”线程中调用了arriveAndDeregister()减少一个 party 之后，后面的线程使用getRegisteredParties()得到的是已经被修改后的 parties 了。但是当前这个阶段(phase)，仍然是需要 4 个 parties 都 arrive 才触发屏障的。从下一个阶段开始，才需要 3 个 parties 都 arrive 就触发屏障。

Phaser 类用来控制某个阶段的线程数量很有用，但它并不在意这个阶段具体有哪些线程 arrive，只要达到它当前阶段的 parties 值，就触发屏障。所以我这里的案例虽然制定了特定的线程（加载新手教程）来更直观地表述 Phaser 的功能，但其实 Phaser 是没有分辨具体是哪个线程的功能的，它在意的只是数量。

它内部使用了两个基于 Fork-Join 框架的原子类辅助：

private final AtomicReference<QNode> evenQ;
private final AtomicReference<QNode> oddQ;

static final class QNode implements ForkJoinPool.ManagedBlocker {
	// 实现代码
}

标签：Phaser,Thread,int,通信,线程,new,工具,public
From： https://www.cnblogs.com/sprinining/p/18326732

相关文章

  • QT mainwindow UI界面添加工具栏
    1.在mainwindowUI设计器界面右上角右键mainwindow 弹出如下菜单图1可以看到添加工具栏，移除状态栏等相关操作都在菜单中2.新建action相关菜单项图2在红框中的ActionEdit中，第一行菜单栏按钮（分别是新建，复制，粘贴，删除，修改）点击以进行创建鼠......
  • Linux性能分析工具Perf
    Linux性能分析工具Perf介绍Perf全名是PerformanceEvent，是在Linux2.6.31以后内建的系统效能分析工具，依靠perf，应用程式可以利用PMU(PerformanceMonitoringUnit),tracepoint和核心内部的特殊计数器(counter)来进行统计，另外还能同时分析运行中的核心程式码，从而更全面了解应用......
  • Ps：选区画笔工具
    选区画笔工具 SelectionBrushTool既可以‍像套索工具‍一样创建选区，也可以像快速蒙版模式一样通过画笔涂抹创建选区。此工具为创建粗略选区带来了极大便利，不仅适合触控板用户，也适合使用生成式填充添加和移除内容。快捷键：L选区画笔工具是Photoshop2024年7月版（25.11......
  • RK3588-添加4G通信(EC20)
    RK3588-Android12添加4G通信文章目录RK3588-Android12添加4G通信1.相关驱动添加---挂载EC20硬件1.1.qmi_wwan_q驱动1.2.option.c添加EC201.3.验证EC20挂载2.添加RIL库2.1.RIL服务声明2.2.配置自动RILD拨号3.设备文件权限和......
  • 使用操作系统工具通过 Python 3 扫描图像
    我正在使用python构建一个应用程序，它需要能够连接到扫描仪并处理生成的pdf/jpeg/其他文件中的数据。我一直在尝试找到一种连接到扫描仪以请求扫描的方法，但没有真正的进展。我已经查看了我能找到的所有选项，包括这个答案：我想连接我的图像扫描仪程序我发现的大多......
  • 类似于 Prython 的工具
    我正在使用R开展一个生物统计学项目。我有几个数据集可供使用。然而，在RStudio环境中，我需要对每个脚本进行一一修改并执行。最近，我发现了一个名为Prython的IDE。它的优点之一是您可以以2D布局查看每个脚本。然而，根据网上的资源来看，用户似乎很少，这使得我将所有工作转移到Pr......
  • 数据仓库建模工具之一——Hive学习第七天
    5、Hive内置函数（分区和排序后面学习hive中的函数）UDF:一对一UDTF：一对多UDAF：多对一https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF--1.查看系统自带函数showfunctions;--2.显示自带的函数的用法descfunctionxxxx;--3.详细显示自带的函数的......
  • OpenAI推出SearchGPT：革新搜索体验的新工具
    引言原文链接在信息爆炸的时代，搜索引擎已经成为人们日常生活中不可或缺的工具。然而，传统的搜索引擎在理解复杂查询和提供准确答案方面仍有许多不足。为了解决这一问题，OpenAI与20240725推出了SearchGPT原型，将生成式AI与传统搜索相结合，为用户提供更智能、更相关的搜索体验。......
  • 思维导图工具MindMap本地docker一键安装详细教程
    文章目录前言1.Docker一键部署思维导图2.本地访问测试3.Linux安装Cpolar4.配置公网地址5.远程访问思维导图6.固定Cpolar公网地址7.固定地址访问前言本文主要介绍在Linux系统以docker方式一键部署思维导图工具SimpleMindMap,并结合cpolar内网穿透工具实现远程......
  • Postman自定义插件全攻略：扩展你的API开发工具箱
    Postman自定义插件全攻略：扩展你的API开发工具箱Postman作为API开发的瑞士军刀，其强大的功能集已经为开发者所熟知。但你知道吗？Postman还允许开发者使用自定义插件来进一步扩展其功能。这些插件可以帮助自动化重复性任务、增强测试能力，甚至集成其他服务。本文将详细介绍如何......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99