日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時(shí)間：8:30-17:00
你可能遇到了下面的問(wèn)題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案
聊聊異步編程的七種實(shí)現(xiàn)方式

大家好，我是 Tom哥

創(chuàng)新互聯(lián)建站于2013年創(chuàng)立，先為自流井等服務(wù)建站，自流井等地企業(yè)，進(jìn)行企業(yè)商務(wù)咨詢服務(wù)。為自流井企業(yè)網(wǎng)站制作PC+手機(jī)+微官網(wǎng)三網(wǎng)同步一站式服務(wù)解決您的所有建站問(wèn)題。

最近有很多小伙伴給我留言，能不能總結(jié)下異步編程，今天就和大家簡(jiǎn)單聊聊這個(gè)話題。

早期的系統(tǒng)是同步的，容易理解，我們來(lái)看個(gè)例子。

同步編程

當(dāng)用戶創(chuàng)建一筆電商交易訂單時(shí)，要經(jīng)歷的業(yè)務(wù)邏輯流程還是很長(zhǎng)的，每一步都要耗費(fèi)一定的時(shí)間，那么整體的RT就會(huì)比較長(zhǎng)。

于是，聰明的人們開(kāi)始思考能不能將一些非核心業(yè)務(wù)從主流程中剝離出來(lái)，于是有了異步編程雛形。

異步編程是讓程序并發(fā)運(yùn)行的一種手段。它允許多個(gè)事件同時(shí)發(fā)生，當(dāng)程序調(diào)用需要長(zhǎng)時(shí)間運(yùn)行的方法時(shí)，它不會(huì)阻塞當(dāng)前的執(zhí)行流程，程序可以繼續(xù)運(yùn)行。

核心思路：采用多線程優(yōu)化性能，將串行操作變成并行操作。異步模式設(shè)計(jì)的程序可以顯著減少線程等待，從而在高吞吐量場(chǎng)景中，極大提升系統(tǒng)的整體性能，顯著降低時(shí)延。

接下來(lái)，我們來(lái)講下異步有哪些編程實(shí)現(xiàn)方式。

一、線程 Thread

直接繼承 Thread類 是創(chuàng)建異步線程最簡(jiǎn)單的方式。

首先，創(chuàng)建Thread子類，普通類或匿名內(nèi)部類方式；然后創(chuàng)建子類實(shí)例；最后通過(guò)start()方法啟動(dòng)線程。

public class AsyncThread extends Thread{
    @Override
    public void run() {
        System.out.println("當(dāng)前線程名稱:" + this.getName() + ", 執(zhí)行線程名稱:" + Thread.currentThread().getName() + "-hello");
    }
}
public static void main(String[] args) {
  // 模擬業(yè)務(wù)流程
  // ....... 
    // 創(chuàng)建異步線程 
    AsyncThread asyncThread = new AsyncThread();
    // 啟動(dòng)異步線程
    asyncThread.start();
}

當(dāng)然如果每次都創(chuàng)建一個(gè) Thread線程，頻繁的創(chuàng)建、銷毀，浪費(fèi)系統(tǒng)資源。我們可以采用線程池。

@Bean(name = "executorService")
public ExecutorService downloadExecutorService() {
    return new ThreadPoolExecutor(20, 40, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2000),
            new ThreadFactoryBuilder().setNameFormat("defaultExecutorService-%d").build(),
            (r, executor) -> log.error("defaultExecutor pool is full! "));
}

將業(yè)務(wù)邏輯封裝到 Runnable 或 Callable 中，交由 線程池 來(lái)執(zhí)行。

二、Future

上述方式雖然達(dá)到了多線程并行處理，但有些業(yè)務(wù)不僅僅要執(zhí)行過(guò)程，還要獲取執(zhí)行結(jié)果。

Java 從1.5版本開(kāi)始，提供了 Callable 和 Future，可以在任務(wù)執(zhí)行完畢之后得到任務(wù)執(zhí)行結(jié)果。

當(dāng)然也提供了其他功能，如：取消任務(wù)、查詢?nèi)蝿?wù)是否完成等。

Future類位于java.util.concurrent包下，接口定義：

public interface Future {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

方法描述：

  • cancel()：取消任務(wù)，如果取消任務(wù)成功返回true，如果取消任務(wù)失敗則返回false。
  • isCancelled()：表示任務(wù)是否被取消成功，如果在任務(wù)正常完成前被取消成功，則返回 true。
  • isDone()：表示任務(wù)是否已經(jīng)完成，如果完成，返回true。
  • get()：獲取執(zhí)行結(jié)果，這個(gè)方法會(huì)產(chǎn)生阻塞，會(huì)一直等到任務(wù)執(zhí)行完畢才返回。
  • get(long timeout, TimeUnit unit)：用來(lái)獲取執(zhí)行結(jié)果，如果在指定時(shí)間內(nèi)，還沒(méi)獲取到結(jié)果，就直接返回null。

代碼示例：

public class CallableAndFuture {
    public static ExecutorService executorService = new ThreadPoolExecutor(4, 40,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue(1024), new ThreadFactoryBuilder()
            .setNameFormat("demo-pool-%d").build(), new ThreadPoolExecutor.AbortPolicy());
    static class MyCallable implements Callable {
        @Override
        public String call() throws Exception {
            return "異步處理，Callable 返回結(jié)果";
        }
    }
    public static void main(String[] args) {
        Future future = executorService.submit(new MyCallable());
        try {
            System.out.println(future.get());
        } catch (Exception e) {
            // nodo
        } finally {
            executorService.shutdown();
        }
    }
}

Future 表示一個(gè)可能還沒(méi)有完成的異步任務(wù)的結(jié)果，通過(guò) ??get?? 方法獲取執(zhí)行結(jié)果，該方法會(huì)阻塞直到任務(wù)返回結(jié)果。

三、FutureTask

FutureTask 實(shí)現(xiàn)了 RunnableFuture 接口，則 RunnableFuture 接口繼承了 Runnable 接口和 Future 接口，所以可以將 FutureTask 對(duì)象作為任務(wù)提交給 ThreadPoolExecutor 去執(zhí)行，也可以直接被 Thread 執(zhí)行；又因?yàn)閷?shí)現(xiàn)了 Future 接口，所以也能用來(lái)獲得任務(wù)的執(zhí)行結(jié)果。

FutureTask 構(gòu)造函數(shù)：

public FutureTask(Callable callable)
public FutureTask(Runnable runnable, V result)

FutureTask 常用來(lái)封裝 Callable 和 Runnable，可以作為一個(gè)任務(wù)提交到線程池中執(zhí)行。除了作為一個(gè)獨(dú)立的類之外，也提供了一些功能性函數(shù)供我們創(chuàng)建自定義 task 類使用。

FutureTask 線程安全由CAS來(lái)保證。

ExecutorService executor = Executors.newCachedThreadPool();
// FutureTask包裝callbale任務(wù)，再交給線程池執(zhí)行
FutureTask futureTask = new FutureTask<>(() -> {
    System.out.println("子線程開(kāi)始計(jì)算：");
    Integer sum = 0;
    for (int i = 1; i <= 100; i++)
        sum += i;
    return sum;
});
// 線程池執(zhí)行任務(wù)， 運(yùn)行結(jié)果在 futureTask 對(duì)象里面
executor.submit(futureTask);
try {
    System.out.println("task運(yùn)行結(jié)果計(jì)算的總和為：" + futureTask.get());
} catch (Exception e) {
    e.printStackTrace();
}
executor.shutdown();

Callable 和 Future 的區(qū)別：Callable 用于產(chǎn)生結(jié)果，F(xiàn)uture 用于獲取結(jié)果。

如果是對(duì)多個(gè)任務(wù)多次自由串行、或并行組合，涉及多個(gè)線程之間同步阻塞獲取結(jié)果，F(xiàn)uture 代碼實(shí)現(xiàn)會(huì)比較繁瑣，需要我們手動(dòng)處理各個(gè)交叉點(diǎn)，很容易出錯(cuò)。

四、異步框架 CompletableFuture

Future 類通過(guò) get() 方法阻塞等待獲取異步執(zhí)行的運(yùn)行結(jié)果，性能比較差。

JDK1.8 中，Java 提供了 CompletableFuture 類，它是基于異步函數(shù)式編程。相對(duì)阻塞式等待返回結(jié)果，CompletableFuture 可以通過(guò)回調(diào)的方式來(lái)處理計(jì)算結(jié)果，實(shí)現(xiàn)了異步非阻塞，性能更優(yōu)。

優(yōu)點(diǎn)：

  • 異步任務(wù)結(jié)束時(shí)，會(huì)自動(dòng)回調(diào)某個(gè)對(duì)象的方法。
  • 異步任務(wù)出錯(cuò)時(shí)，會(huì)自動(dòng)回調(diào)某個(gè)對(duì)象的方法。
  • 主線程設(shè)置好回調(diào)后，不再關(guān)心異步任務(wù)的執(zhí)行。

泡茶示例：

(內(nèi)容摘自：極客時(shí)間的《Java 并發(fā)編程實(shí)戰(zhàn)》)。

//任務(wù)1：洗水壺->燒開(kāi)水
CompletableFuture f1 =
        CompletableFuture.runAsync(() -> {
            System.out.println("T1:洗水壺...");
            sleep(1, TimeUnit.SECONDS);

            System.out.println("T1:燒開(kāi)水...");
            sleep(15, TimeUnit.SECONDS);
        });
//任務(wù)2：洗茶壺->洗茶杯->拿茶葉
CompletableFuture f2 =
        CompletableFuture.supplyAsync(() -> {
            System.out.println("T2:洗茶壺...");
            sleep(1, TimeUnit.SECONDS);

            System.out.println("T2:洗茶杯...");
            sleep(2, TimeUnit.SECONDS);

            System.out.println("T2:拿茶葉...");
            sleep(1, TimeUnit.SECONDS);
            return "龍井";
        });

//任務(wù)3：任務(wù)1和任務(wù)2完成后執(zhí)行：泡茶
CompletableFuture f3 =
        f1.thenCombine(f2, (__, tf) -> {
            System.out.println("T1:拿到茶葉:" + tf);
            System.out.println("T1:泡茶...");
            return "上茶:" + tf;
        });
//等待任務(wù)3執(zhí)行結(jié)果
System.out.println(f3.join());

}

CompletableFuture 提供了非常豐富的API，大約有50種處理串行，并行，組合以及處理錯(cuò)誤的方法。

五、 SpringBoot 注解 @Async

除了硬編碼的異步編程處理方式，SpringBoot 框架還提供了 注解式 解決方案，以 方法體 為邊界，方法體內(nèi)部的代碼邏輯全部按異步方式執(zhí)行。

首先，使用 @EnableAsync 啟用異步注解。

@SpringBootApplication
@EnableAsync
public class StartApplication {
    public static void main(String[] args) {
        SpringApplication.run(StartApplication.class, args);
    }
}

自定義線程池：

@Configuration
@Slf4j
public class ThreadPoolConfiguration {

    @Bean(name = "defaultThreadPoolExecutor", destroyMethod = "shutdown")
    public ThreadPoolExecutor systemCheckPoolExecutorService() {

        return new ThreadPoolExecutor(3, 10, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue(10000),
                new ThreadFactoryBuilder().setNameFormat("default-executor-%d").build(),
                (r, executor) -> log.error("system pool is full! "));
    }
}

在異步處理的方法上添加注解 @Async ，當(dāng)對(duì) execute 方法 調(diào)用時(shí)，通過(guò)自定義的線程池 defaultThreadPoolExecutor 異步化執(zhí)行  execute 方法。

@Service
public class AsyncServiceImpl implements AsyncService {
    @Async("defaultThreadPoolExecutor")
    public Boolean execute(Integer num) {
        System.out.println("線程：" + Thread.currentThread().getName() + " , 任務(wù)：" + num);
        return true;
    }
}

用 @Async 注解標(biāo)記的方法，稱為異步方法。在spring boot應(yīng)用中使用 @Async 很簡(jiǎn)單：

  • 調(diào)用異步方法類上或者啟動(dòng)類加上注解 @EnableAsync。
  • 在需要被異步調(diào)用的方法外加上 @Async。
  • 所使用的 @Async 注解方法的類對(duì)象應(yīng)該是Spring容器管理的bean對(duì)象。

六、Spring ApplicationEvent 事件

事件機(jī)制在一些大型項(xiàng)目中被經(jīng)常使用，Spring 專門提供了一套事件機(jī)制的接口，滿足了架構(gòu)原則上的解耦。

ApplicationContext 通過(guò) ApplicationEvent 類和 ApplicationListener 接口進(jìn)行事件處理。如果將實(shí)現(xiàn) ApplicationListener 接口的 bean 注入到上下文中，則每次使用 ApplicationContext 發(fā)布 ApplicationEvent 時(shí)，都會(huì)通知該 bean。本質(zhì)上，這是標(biāo)準(zhǔn)的觀察者設(shè)計(jì)模式。

ApplicationEvent 是由 Spring 提供的所有 Event 類的基類。

首先，自定義業(yè)務(wù)事件子類，繼承自 ApplicationEvent，通過(guò)泛型注入業(yè)務(wù)模型參數(shù)類。相當(dāng)于 MQ 的消息體。

public class OrderEvent extends AbstractGenericEvent {
    public OrderEvent(OrderModel source) {
        super(source);
    }
}

然后，編寫事件監(jiān)聽(tīng)器。ApplicationListener 接口是由 Spring 提供的事件訂閱者必須實(shí)現(xiàn)的接口，我們需要定義一個(gè)子類，繼承 ApplicationListener。相當(dāng)于 MQ 的消費(fèi)端。

@Component
public class OrderEventListener implements ApplicationListener {
    @Override
    public void onApplicationEvent(OrderEvent event) {
        System.out.println("【OrderEventListener】監(jiān)聽(tīng)器處理！" + JSON.toJSONString(event.getSource()));
    }
}

最后，發(fā)布事件，把某個(gè)事件告訴所有與這個(gè)事件相關(guān)的監(jiān)聽(tīng)器。相當(dāng)于 MQ 的生產(chǎn)端。

OrderModel orderModel = new OrderModel();
orderModel.setOrderId((long) i);
orderModel.setBuyerName("Tom-" + i);
orderModel.setSellerName("judy-" + i);
orderModel.setAmount(100L);
// 發(fā)布Spring事件通知
SpringUtils.getApplicationContext().publishEvent(new OrderEvent(orderModel));

加個(gè)餐：

[消費(fèi)端]線程：http-nio-8090-exec-1，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-1","orderId":1,"sellerName":"judy-1"}
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 1
[消費(fèi)端]線程：http-nio-8090-exec-1，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-2","orderId":2,"sellerName":"judy-2"}
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 2
[消費(fèi)端]線程：http-nio-8090-exec-1，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-3","orderId":3,"sellerName":"judy-3"}
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 3

上面是跑了個(gè)demo的運(yùn)行結(jié)果，我們發(fā)現(xiàn)無(wú)論生產(chǎn)端還是消費(fèi)端，使用了同一個(gè)線程 http-nio-8090-exec-1，Spring 框架的事件機(jī)制默認(rèn)是同步阻塞的。只是在代碼規(guī)范方面做了解耦，有較好的擴(kuò)展性，但底層還是采用同步調(diào)用方式。

那么問(wèn)題來(lái)了，如果想實(shí)現(xiàn)異步調(diào)用，如何處理？

我們需要手動(dòng)創(chuàng)建一個(gè) SimpleApplicationEventMulticaster，并設(shè)置 TaskExecutor，此時(shí)所有的消費(fèi)事件采用異步線程執(zhí)行。

@Component
public class SpringConfiguration {
    @Bean
    public SimpleApplicationEventMulticaster applicationEventMulticaster(@Qualifier("defaultThreadPoolExecutor") ThreadPoolExecutor defaultThreadPoolExecutor) {
        SimpleApplicationEventMulticaster simpleApplicationEventMulticaster = new SimpleApplicationEventMulticaster();
        simpleApplicationEventMulticaster.setTaskExecutor(defaultThreadPoolExecutor);
        return simpleApplicationEventMulticaster;
    }
}

我們看下改造后的運(yùn)行結(jié)果：

[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 1
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 2
[生產(chǎn)端]線程：http-nio-8090-exec-1，發(fā)布事件 3
[消費(fèi)端]線程：default-executor-1，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-2","orderId":2,"sellerName":"judy-2"}
[消費(fèi)端]線程：default-executor-2，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-1","orderId":1,"sellerName":"judy-1"}
[消費(fèi)端]線程：default-executor-0，消費(fèi)事件 {"amount":100.0,"buyerName":"Tom-3","orderId":3,"sellerName":"judy-3"}

SimpleApplicationEventMulticaster 這個(gè)我們自己實(shí)例化的 Bean 與系統(tǒng)默認(rèn)的加載順序如何？會(huì)不會(huì)有沖突？

查了下 Spring 源碼，處理邏輯在 AbstractApplicationContext#initApplicationEventMulticaster 方法中，通過(guò) beanFactory 查找是否有自定義的 Bean，如果沒(méi)有，容器會(huì)自己 new 一個(gè) SimpleApplicationEventMulticaster 對(duì)象注入到容器中。

代碼地址：https://github.com/aalansehaiyang/wx-project。

七、消息隊(duì)列

異步架構(gòu)是互聯(lián)網(wǎng)系統(tǒng)中一種典型架構(gòu)模式，與同步架構(gòu)相對(duì)應(yīng)。而消息隊(duì)列天生就是這種異步架構(gòu)，具有超高吞吐量和超低時(shí)延。

消息隊(duì)列異步架構(gòu)的主要角色包括消息生產(chǎn)者、消息隊(duì)列和消息消費(fèi)者。

消息生產(chǎn)者就是主應(yīng)用程序，生產(chǎn)者將調(diào)用請(qǐng)求封裝成消息發(fā)送給消息隊(duì)列。

消息隊(duì)列的職責(zé)就是緩沖消息，等待消費(fèi)者消費(fèi)。根據(jù)消費(fèi)方式又分為點(diǎn)對(duì)點(diǎn)模式和發(fā)布訂閱模式兩種。

消息消費(fèi)者，用來(lái)從消息隊(duì)列中拉取、消費(fèi)消息，完成業(yè)務(wù)邏輯處理。

當(dāng)然市面上消息隊(duì)列框架非常多，常見(jiàn)的有RabbitMQ、Kafka、RocketMQ、ActiveMQ 和 Pulsar 等。

不同的消息隊(duì)列的功能特性會(huì)略有不同，但整體架構(gòu)類似，這里就不展開(kāi)了。

我們只需要記住一個(gè)關(guān)鍵點(diǎn)，借助消息隊(duì)列這個(gè)中間件可以高效的實(shí)現(xiàn)異步編程。


分享標(biāo)題：聊聊異步編程的七種實(shí)現(xiàn)方式
文章轉(zhuǎn)載：http://www.dlmjj.cn/article/cojdgse.html