首页 / JAVA / java线程异步转同步。

java线程异步转同步。

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了java线程异步转同步。，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含10092字，纯文字阅读大概需要15分钟。

内容图文

最近有个项目在压测，TPS有点低。做了一些日志异步批量落地和redis数据预热后，TPS稍微提高了点，但还是没达标。

研究了下项目的系统设计和技术栈。

用的分布式服务架构，其中服务端使用Netty。而客户端为了同步获取响应结果，采用了socket短连接模式。

为了使系统的性能有所提升，决定客户端也改用Netty框架并采用长连接的方式。

Netty是Jboss开源的一款非常优秀的异步通信框架。目前很多主流的开源项目有使用Netty开发的，Dubbo/RocketMQ/Apache Synapse等。

但Netty本身似乎没有提供同步等待响应的接口或方法。

使用过Dubbo的都知道我们平常使用的dubbo发布/订阅端就是同步获取响应报文，虽然它本身是基于Netty开发。

为了知道dubbo怎么实现netty响应结果异步转同步，看了点dubbo的源码。

发现dubbo是使用future+lock+condition实现的。这里就不作展开了，有兴趣可以看一下Dubbo的DefaultFuture这个类。

这里记录一下网上看到的一些异步转同步的方法。

首先定义一个业务操作类，纯粹处理业务。

/**
 * 业务操作类
 */
public class TaskService {

    public String getNumber() {
        return UUID.randomUUID().toString();
    }
}

然后写一个线程执行接口，

public interface Executor {

    /**
     * 异步执行
     */
    default void supplyAsync() {}

    /**
     * 异步执行并回调
     * @param callback
     */
    default void supplyAsync(Callback callback) {}

    /**
     * 同步执行
     */
    default void supplySync(){}

    /**
     * 回调接口
     */
    @FunctionalInterface
    public interface Callback{
        void call(Object o);
    }
}

接下来，写异步操作。

/**
 * 异步任务
 */
public class AsyncExecutor implements Executor {

    private TaskService service = new TaskService();

    public void supplyAsync() {
        this.supplyAsync(null);
    }

    public void supplyAsync(Callback callback) {
        // 另起线程异步执行。
        new Thread(() -> {
            System.out.println("running async task...");
            try {
                // 子线程睡眠，主线程不休息。
                Thread.sleep(2000);
            } catch (InterruptedException e) {

            }
            String num = service.getNumber();
            // 如果有回调，则调用回调函数。
            if (callback != null){
                callback.call(num);
            }
        }).start();
    }
}

测试代码：

public static void main(String[] args) throws Exception {
        System.err.println("\n===============   start main thread   ===============\n");
        Executor executor;
        /**
         *  async
         */
        executor = new AsyncExecutor();
        executor.supplyAsync((o) -> {
            System.out.println(String.format("get number[%s] by async.", o));
        });
        System.err.println("\n===============   main thread  over   ===============\n");
    }

输出：

Connected to the target VM, address: '127.0.0.1:53015', transport: 'socket'

===============   start main thread   ===============


===============   main thread  over   ===============

running async task...
get number[de720331-2ca3-4b33-8dee-4a7d26ede037] by async.
Disconnected from the target VM, address: '127.0.0.1:53015', transport: 'socket'

我们可以看到，子线程在睡眠了2秒的时候，主线程并没有等待子线程执行完，而是继续往下执行。

接下来，我们要将异步响应改成同步的。

/**
 * 同步执行抽象类
 */
public abstract class AbstractSyncExecutor implements Executor {
    // 异步执行器
    public AsyncExecutor executor = new AsyncExecutor();

    /**
     * 异步转同步
     */
    public void supplySync() {
        executor.supplyAsync(this::process);
        await();
    }

    /**
     *  线程等待
     */
    protected void await() {}

    /**
     * 回调
     * @param o
     */
    protected abstract void process(Object o);
}

第一种：Synchronized + wait + notify

/**
 * 采用synchronized配合wait和notify。
 */
public class SynchronizedExecutor extends AbstractSyncExecutor {

    @Override
    protected void process(Object o) {
        System.out.println(String.format("get number[%s] by synchronized.", o));
        synchronized (this) {
            notify();
        }
    }

    @Override
    public void await() {
        synchronized (this) {
            try {
                // 主线程调用wait阻塞等待，直到回调方法调用notify或者notifyAll唤醒。
                wait();
            } catch (InterruptedException e) {

            }
        }
    }
}

测试代码

public static void main(String[] args) throws Exception {
        System.err.println("\n===============   start main thread   ===============\n");
        Executor executor;

        /**
         *  synchronized
         */
        executor = new SynchronizedExecutor();
        executor.supplySync();

        System.err.println("\n===============   main thread  over   ===============\n");
    }

执行结果

Connected to the target VM, address: '127.0.0.1:53235', transport: 'socket'

===============   start main thread   ===============

running async task...
get number[05c5db70-ec3b-4411-b080-21e5b4cddf79] by synchronized.

===============   main thread  over   ===============

Disconnected from the target VM, address: '127.0.0.1:53235', transport: 'socket'

可以看到内容已经顺序输出了。

第二种：reentrantLock + condition

/**
 * 使用lock + condition
 */
public class ReentrantLockExecutor extends AbstractSyncExecutor {

    private Lock lock = new ReentrantLock();
    private Condition condition;

    public ReentrantLockExecutor() {
        this.condition = lock.newCondition();
    }

    @Override
    protected void process(Object o) {
        System.out.println(String.format("get number[%s] by lockAndCondition.", o));
        lock.lock();
        try {
            condition.signal();
        } finally {
            lock.unlock();
        }
    }

    @Override
    protected void await() {
        lock.lock();
        try {
            // 阻塞等待直到回调函数唤醒
            condition.await();
        } catch (Exception e) {

        } finally {
            lock.unlock();
        }
    }
}

测试代码

    public static void main(String[] args) throws Exception {
        System.err.println("\n===============   start main thread   ===============\n");
        Executor executor;

        /**
         *  reentrantLock
         */
        executor = new ReentrantLockExecutor();
        executor.supplySync();

        System.err.println("\n===============   main thread  over   ===============\n");
    }

执行结果

Connected to the target VM, address: '127.0.0.1:53254', transport: 'socket'

===============   start main thread   ===============

running async task...
get number[6ad17c2c-a3ce-4fc6-a03a-fdb7d84acf7e] by lockAndCondition.

===============   main thread  over   ===============

Disconnected from the target VM, address: '127.0.0.1:53254', transport: 'socket'

第三种：countDownLatch

/**
 * CountDownLatch
 */
public class CountDownLatchExecutor extends AbstractSyncExecutor {

    // 假设每笔调用都创建一个CountDownLatchExecutor，那么从发起到响应只算一次操作，这里设置为1就可以了。
    private CountDownLatch latch = new CountDownLatch(1);

    @Override
    public void process(Object o) {
        System.out.println(String.format("get number[%s] by countDownLatch.", o));
        // latch count - 1  变成0， 主线程继续执行
        latch.countDown();
    }

    @Override
    protected void await() {
        try {
            // 阻塞直到latch count=0
            latch.await();
        } catch (InterruptedException e) {
            latch.countDown();
        }
    }
}

测试代码

    public static void main(String[] args) throws Exception {
        System.err.println("\n===============   start main thread   ===============\n");
        Executor executor;

        /**
         * countDownLatch
         */
        executor = new CountDownLatchExecutor();
        executor.supplySync();

        System.err.println("\n===============   main thread  over   ===============\n");
    }

执行结果

Connected to the target VM, address: '127.0.0.1:53325', transport: 'socket'

===============   start main thread   ===============

running async task...
get number[aa51089e-6f74-4ae7-b463-6a04ca73adcf] by countDownLatch.

===============   main thread  over   ===============

Disconnected from the target VM, address: '127.0.0.1:53325', transport: 'socket'

第四种：CyclicBarrier

/**
 * CyclicBarrier
 */
public class CyclicBarrierExecutor extends AbstractSyncExecutor {
    
    // 假设每笔调用都创建一个CountDownLatchExecutor
    CyclicBarrier barrier = new CyclicBarrier(2);

    @Override
    protected void process(Object o) {
        try {
            System.out.println(String.format("get number[%s] by cyclicBarrier.", o));
            // await线程数量=2，当前线程被唤醒
            barrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void await() {
        try {
            // await线程数为1，等待直至所有线程都到达
            barrier.await();
        } catch (Exception e) {

        }
    }
}

与CountDownLatch相反。CyclicBarrier是做加操作。当await线程达到初始parties数时，当前线程就被唤醒。我们需要在主线程await一次，回调线程await一次，然后主线程唤醒。即：CyclicBarrier的栅栏数parties设置为2。

测试代码

    public static void main(String[] args) throws Exception {
        System.err.println("\n===============   start main thread   ===============\n");
        Executor executor;

        /**
         * CyclicBarrier
         */
        executor = new CyclicBarrierExecutor();
        executor.supplySync();

        System.err.println("\n===============   main thread  over   ===============\n");
    }

执行结果

Connected to the target VM, address: '127.0.0.1:53527', transport: 'socket'

===============   start main thread   ===============

running async task...
get number[2409d6c4-7bb2-4838-a6ef-3ce42e4a18c7] by cyclicBarrier.

===============   main thread  over   ===============

Disconnected from the target VM, address: '127.0.0.1:53527', transport: 'socket'

第五种：Future + countDownLatch。

public class SyncFuture<T> implements Future<T> {
    private CountDownLatch latch = new CountDownLatch(1);
    private T resp;

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        if (this.resp != null) {
            return true;
        }
        return false;
    }

    @Override
    public T get() throws InterruptedException, ExecutionException {
        latch.await();
        return this.resp;
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (latch.await(timeout, unit)) {
            return this.resp;
        }
        return null;
    }

    public void set(T resp) {
        this.resp = resp;
        latch.countDown();
    }
}

public class FutureExecutor extends AbstractSyncExecutor {
    private SyncFuture future;

    public FutureExecutor(SyncFuture future) {
        this.future = future;
    }

    @Override
    public void process(Object o) {
        future.set(o);
    }
}

测试代码

    public static void main(String[] args) throws Exception {
        System.err.println("\n===============   start main thread   ===============\n");

        /**
         * future + countDownLatch
         */
        SyncFuture<String> future = new SyncFuture<>();
        new FutureExecutor(future).supplySync();
        // Object resp = futureExecutor.get();
        // Object resp = futureExecutor.get(1, TimeUnit.SECONDS);
        Object resp = future.get(3, TimeUnit.SECONDS);
        System.out.println(String.format("get number[%s] by futureAndCountDownLatch.", resp));

        System.err.println("\n===============   main thread  over   ===============\n");
    }

执行结果

Connected to the target VM, address: '127.0.0.1:53590', transport: 'socket'

===============   start main thread   ===============

running async task...
get number[942e6c15-1eb2-4572-9a1f-dcb7d262387b] by futureAndCountDownLatch.

===============   main thread  over   ===============

Disconnected from the target VM, address: '127.0.0.1:53590', transport: 'socket'

前面四种方式都大同小异。主线程阻塞等待，子线程的回调函数里面唤醒主线程。

只有最后一种不太一样。而Dubbo就是使用最后一种方式，只是把其中的countDownLatch换成了condition+lock。

内容总结

以上是互联网集市为您收集整理的java线程异步转同步。全部内容，希望文章能够帮你解决java线程异步转同步。所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/630732.html

来源：【匿名】

【上一篇】java 基础知识【下一篇】JAVA/JSP学习系列之五

更多 ►

【java线程异步转同步。】教程文章相关的互联网学习教程文章

深入理解Java虚拟机- 学习笔记 - Java内存模型与线程

除了在硬件上增加告诉缓存之外，为了使得处理器内部的运算单元能尽量被充分利用，处理器可能会对输入代码进行乱序执行(Out-Of-Order Execution)优化，处理器会在计算之后将乱序执行的结果重组，保证该结果与顺序执行的结果一致，但并不保证程序中各个语句计算的先后顺序与输入代码中的顺序一致，因此，如果存在一个计算任务依赖另外一个计算任务的中间结果，那么其顺序性并不能靠代码的先后顺序来保证。与处理器的乱序优化执行类似...

Java 如何判断线程池所有任务是否执行完毕【代码】

shutdownvoid shutdown()启动一次顺序关闭，执行以前提交的任务，但不接受新任务。如果已经关闭，则调用没有其他作用。抛出：SecurityException - 如果安全管理器存在并且关闭，此 ExecutorService 可能操作某些不允许调用者修改的线程（因为它没有保持RuntimePermission("modifyThread")），或者安全管理器的 checkAccess 方法拒绝访问。isTerminatedboolean isTerminated()如果关闭后所有任务都已完成，则返回 true。注意，除非首...

java多线程 21 ： ReentrantReadWriteLock ，synchronized和ReentrantLock的对比【代码】

读写锁ReentrantReadWriteLock概述大型网站中很重要的一块内容就是数据的读写，ReentrantLock虽然具有完全互斥排他的效果（即同一时间只有一个线程正在执行lock后面的任务），但是效率非常低。所以在JDK中提供了一种读写锁ReentrantReadWriteLock，使用它可以加快运行效率。读写锁表示两个锁，一个是读操作相关的锁，称为共享锁；另一个是写操作相关的锁，称为排他锁。我把这两个操作理解为三句话：1、读和读之间不互斥，因为读操作...

JAVA多线程（七）模式-Producer Consumer【代码】

Producer Consumer生产者创建数据，通过中介控制流量并安全传递给消费者。适用环境生产者生产数据的速度与消费者处理数据的速度不一致，中介者通过缓存和阻塞对消费者的数据压力进行调整。样例4生产者生产产品，放入市场，2消费者消费。产品package ProducerConsumer;public class Product {private String prdId=null;public Product(String prdId) {this.prdId=prdId;}public String getOrderId(){return this.prdId;} }市场pack...

Java多线程:线程同步与关键字synchronized

一、同步的特性1. 不必同步类中所有的方法, 类可以同时拥有同步和非同步方法.2. 如果线程拥有同步和非同步方法, 则非同步方法可以被多个线程自由访问而不受锁的限制. 参见实验1:http://blog.csdn.net/huang_xw/article/details/73185613. 如果两个线程要执行一个类中的同步方法, 并且两个线程使用相同的实例来调用方法, 那么一次只能有一个线程能够执行方法, 另一个需要等待, 直到锁被释放. 参见实验2:http://blog.csdn.net/huang_...

Java线程核心基础（上）【代码】【图】

Java线程核心基础（上）一、实现多线程　　根据Oracle官方文档，目前推荐的创建线程方法主要有两种，分别是继承Thread类和实现Runnable接口。通过阅读Thread类源码，可以发现继承Thread类需要重写run()方法，而实现Runnable接口会将自己实现的对象在new Thread()时，通过Thread构造函数传给Thread类中的target对象，并在调用run()方法时调用target.run()，下面让我们看源码。/* What will be run. 这是Thread类中的target对象 */ p...

JAVA多线程之wait/notify【代码】【图】

本文主要学习JAVA多线程中的 wait()方法与 notify()/notifyAll()方法的用法。①wait() 与 notify/notifyAll 方法必须在同步代码块中使用②wait() 与 notify/notifyAll() 的执行过程③中断调用wait()方法进入等待队列的线程④notify 通知的顺序不能错⑤多线程中测试某个条件的变化用 if 还是用 while？ ①wait() 与 notify/notifyAll 方法必须在同步代码块中使用wait() 与 notify/notifyAll() 是Object类的方法，在执行两个方法...

使用java.util.concurrent.ThreadFactory类创建线程

工厂设计模式是Java中最常用的设计模式之一。它是一种创建型设计模式，能够用于创建一个或多个类所需要的对象。有了这个工厂，我们就能集中的创建对象。集中创建方式给我们带来了一些好处，例如：1. 能够很容易的改变类创建的对象或者创建对象的方式；2. 能够很容易限制对象的创建，例如：我们只能为a类创建N个对象；3. 能够很容易的生成有关对象创建的统计数据。在Java中，我们通常使用两种方式来创建线程：继承Thread类和实现Run...

通过 Javacore 了解线程运行状况【代码】【图】

Javacore 是一个当前 JVM 运行状态的快照。通过对Javacore 的分析，可以了解在 JVM 中运行的应用程序的当前状态，比如是否“卡”在某一点上，或在某些代码上运行时间太长。Javacore 的基本内容　　Javacore，也可以称为“threaddump”或是“javadump”，它是 Java 提供的一种诊断特性，能够提供一份可读的当前运行的 JVM 中线程使用情况的快照。即在某个特定时刻，JVM 中有哪些线程在运行，每个线程执行到哪一个类，哪一个方法。 ...

java 线程协作 join()【代码】

在实际开发中我们往往会遇到这样的情况一个线程的执行需要依赖另一个线程执行后的结果。即主线程生成并起动了子线程，如果子线程里要进行大量的耗时的运算，主线程往往将于子线程之前结束，但是如果主线程处理完其他的事务后，需要用到子线程的处理结果，也就是主线程需要等待子线程执行完成之后再结束，这个时候就要用到join()方法了。join()join() 方法主要是让调用该方法的thread完成run方法里面的任务后，再执行join()方法后面...

【JAVA】HashMap的原理及多线程下死循环的原因【图】

再次翻到以前工作中遇到的一个问题，HashMap在多线程下会出现死循环的问题，以前只是知道会死循环，导致CPU100%把机器拖跨，今天来彻底看看首先来看下，HashMap的原理：HashMap是一个数组，对key使用hash算法计算出数组对应的下标i，然后把<key, value>插到table[i]，如果两个不同的key被算在同一个i，那就出现冲突，又叫碰撞，这样就会在table[i]上形成一个链表；总结下来HashMap是一个数组+链表组成的数据结构；我们知道，在往...

Java7并发编程实战（一）线程的中断【代码】

控制线程中断的方法一般常规是定义一个布尔值，然后while（布尔值）去执行，当想停止该线程时候，把布尔值设为false.这里我们来看第二种，Interrupt 该例子模拟一个线程从1打印到10，然后到5的时候中断线程，主要在线程中捕捉InterruptedException 异常 publicclass Test implements Runnable {@Overridepublicvoid run() {// TODO Auto-generated method stubfor(int i=0;i<10;i++){System.out.println("打印"+i);try {TimeUnit...

关于java多线程死锁的实验【代码】

先上完整的代码后续再依次修改代码块得到相应的结论 1//尝试写一个死锁感受一下 2publicclass DeadLock {3publicstaticvoid main(String[] args) {4 Object o1 = new Object();5 Object o2 = new Object();6//创建两个object对象7//创建两个Thread线程 8 Thread T1 = new Thread(new process1(o1,o2)); //两个线程T1、T2共享了相同的两个对象o1、o2 9 Thread T2 = new Thread(new process2(o1,o2)); 10 T1....

十七、Java从头开始-多线程编程：聊天室升级版

多线程编程：聊天室升级版?????? 还记得上一章的聊天室吗，不如说是单向发送器，我只能单方面的发送信息给客户端，不能接收消息，不能看到其他室友的发送的信息，我想要的是一个真正的聊天室，在发送信息的同时能看到我和其他室友发送的聊天记录，这里就不得不一边发送一边接收，意味着在用户与GUI交互的同时不能打断它，需要一个单独的执行空间来执行接收功能，就必须引进一个新的线程（Thread）。一、创建线程????? 创建一个线程...

java程序中的多线程(转)【代码】

为什么会排队等待？下面的这个简单的 Java 程序完成四项不相关的任务。这样的程序有单个控制线程，控制在这四个任务之间线性地移动。此外，因为所需的资源 ― 打印机、磁盘、数据库和显示屏 -- 由于硬件和软件的限制都有内在的潜伏时间，所以每项任务都包含明显的等待时间。因此，程序在访问数据库之前必须等待打印机完成打印文件的任务，等等。如果您正在等待程序的完成，则这是对计算资源和您的时间的一种拙劣使用。改进此程序的...

首页 / JAVA / java线程异步转同步。

java线程异步转同步。

内容导读

内容图文

内容总结

内容备注

内容手机端

【java线程异步转同步。】教程文章相关的互联网学习教程文章

深入理解Java虚拟机- 学习笔记 - Java内存模型与线程

Java 如何判断线程池所有任务是否执行完毕【代码】

java多线程 21 ： ReentrantReadWriteLock ，synchronized和ReentrantLock的对比【代码】

JAVA多线程（七）模式-Producer Consumer【代码】

Java多线程:线程同步与关键字synchronized

Java线程核心基础（上）【代码】【图】

JAVA多线程之wait/notify【代码】【图】

使用java.util.concurrent.ThreadFactory类创建线程

通过 Javacore 了解线程运行状况【代码】【图】

java 线程协作 join()【代码】

【JAVA】HashMap的原理及多线程下死循环的原因【图】

Java7并发编程实战（一）线程的中断【代码】

关于java多线程死锁的实验【代码】

十七、Java从头开始-多线程编程：聊天室升级版

java程序中的多线程(转)【代码】

JAVA - 相关标签

同步 - 相关标签

异步 - 相关标签

JAVA - 技术教程分类

JAVA - 最新教程

JAVA - 最热教程