Asynchronous callback future mode of concurrent mode

Future Pattern （ Get results asynchronously , Self realization ）

The traditional serial process is as follows ：

Waiting will be blocked when getting data , Get the data and then perform other tasks .
and Future The mode will immediately return a voucher （Future）, At this time, you can perform other tasks ; Wait for the data to pass through the previous Future Just get the data from the voucher , The flow is as follows: ：

Let's implement a simple Future Pattern ：
First, analyze and design the following objects ：

• Main The system starts , call Client Request , Get immediate return FutureData
• Client client , Send to get Data Request , Return immediately FutureData, And start thread assembly RealData
• Data Interface to return data
• FutureData Virtual data , It's a voucher , Assembly required RealData
• RealData Real data , Slow build
  The code is as follows ：
  Data Interface ：

public interface Data {
    String getResult() throws InterruptedException;
}

RealData class ：

public class RealData implements Data{
    private String content;
    public RealData(String content){
        this.content = content;
    }
    @Override
    public String getResult() {
        return content;
    }
}

FutureData class ：

public class FutureData implements Data {
    //  Are you ready to 
    private boolean isReady;
    //  Assemble real data 
    private RealData realData;
    @Override
    public synchronized String getResult() {
        //  Not ready , Then block and wait 
        while (!isReady){
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return realData.getResult();
    }
    public synchronized void setRealData(RealData realData){
        if(isReady){
            return;
        }
        this.realData = realData;
        this.isReady = true;
        //  Notify other threads 
        notifyAll();
    }
}

Client class ：

public class Client {
    //  Request data 
    public FutureData request(String queryStr){
        FutureData futureData = new FutureData();
        //  Start the thread to assemble real data asynchronously 
        new Thread(()->{
            //  The long-running 
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //  Build real data 
            RealData realData = new RealData("hello future," + queryStr);
            futureData.setRealData(realData);
        }).start();
        //  Return immediately 
        return futureData;
    }
}

Main Procedure class ：

public class Main {
    public static void main(String[] args) {
        Client client = new Client();
        FutureData futureData = client.request(" test ");
        //  Do other things 
        System.out.println(" Do something else 1");
        System.out.println(" Do something else 2");
        //  To get the results 
        String result = futureData.getResult();
        System.out.println(" The asynchronous result is ："+result);
    }
}

 Output results ：
 Do something else 1
 Do something else 2
 The asynchronous result is ：hello future, test 

JDK Medium Future Pattern

Here's the picture , yes JDK1.8 The encapsulation of Future Pattern realization .(1.5 It introduces Future Pattern ,1.8 Is more powerful , Provides CompletableFuture)

among ：

• class MyCallable Realized Callable Interface call() Method will return real data （ Similar to self implementation Future Data interface in pattern Data Of getResult() Method )
• FutureTask Be similar to FutureData, Are used as immediate return credentials in asynchronous calls
• Thread pools are similar to Client, Are used to perform tasks
  The following is JDK Of future Example ：
  MyCallable class ：

public class MyCallable implements Callable<String> {
    private String str;
    public MyCallable(String str){
        this.str = str;
    }
    @Override
    public String call() throws Exception {
        //  Simulate time-consuming operations 
        Thread.sleep(2000);
        return "hello jdk future," + str;
    }
}

Main Program :

public class Main {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(1);
        FutureTask<String> futureTask = new FutureTask<>(new MyCallable(" test "));
        //  Submit tasks 
        executor.submit(futureTask);
        System.out.println(" Deal with other logic 1");
        System.out.println(" Deal with other logic 2");
        String result = futureTask.get();
        System.out.println("(jdk) The result of asynchronous processing is :"+result);
        executor.shutdown();
    }
}

 Data results ：
 Deal with other logic 1
 Deal with other logic 2
(jdk) The result of asynchronous processing is :hello jdk future, test 

in addition ,JDK Of Future Mode Future The interface also provides some advanced functions .

boolean cancel(boolean mayInterruptIfRunning);//  Cancel the task 
boolean isCanclled();//  Whether it has been cancelled 
boolean isDone();//  Has it been completed 
V get(long timeout,TimeUnit unit);//  Get results within the timeout 

Guava Expand Future Pattern

JDK Of Future In the pattern ,future.get() It's blocked , It is not conducive to high concurrency development .Guava Enhanced Future Pattern , Added callback interface at completion , send future When it is finished, the application can be automatically notified for acquisition processing .

Change the above program to guava How to set the callback function , The code is as follows ：
Main Program ：

public class Main {
    public static void main(String[] args){
        ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(4));
        //  Submit tasks 
        ListenableFuture<String> future = executorService.submit(new MyCallable(" test "));
        //  Add callback function 
        future.addListener(()->{
            String result = null;
            try {
                result = future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            System.out.println("(guava) The result of asynchronous processing is :"+result);
        },MoreExecutors.directExecutor());
        System.out.println(" Deal with other logic 1");
        System.out.println(" Deal with other logic 2");
        executorService.shutdown();
    }
}

Output results ：

 Deal with other logic 1
 Deal with other logic 2
(guava) The result of asynchronous processing is :hello guava future, test 

Netty Expand Future Pattern

netty Support settings are also provided in Future Extension of callback .

public class Main {
    public static void main(String[] args) {
        //  establish netty Thread group 
        EventExecutorGroup group = new DefaultEventExecutorGroup(4);
        //  Submit tasks 
        Future<String> future = group.submit(new MyCallable(" test "));
        System.out.println(" Deal with other logic 1");
        System.out.println(" Deal with other logic 2");
        future.addListener(new FutureListener<String>(){
            @Override
            public void operationComplete(Future<String> future) throws Exception {
                String result = future.get();
                System.out.println("(netty) The result of asynchronous processing is :"+result);
            }
        });
        group.shutdownGracefully();
    }
}

JDK8 Of CompletableFuture

JDK8 Provided in CompletableFuture More powerful .
CompletableFuture Realized CompletionStage Interface and Future Interface , The former is an extension of the latter , Added asynchronous callback 、 Streaming 、 Multiple Future The ability to combine processing , send Java It is more smooth and convenient when dealing with multi task collaborative work .
Use CompletableFuture The transformation supports asynchronous callback methods , The code is as follows ：
MySupplier class ：

public class MySupplier implements Supplier<String> {
    private String str;
    public MySupplier(String str){
        this.str = str;
    }

    @Override
    public String get() {
        //  Simulate time-consuming operations 
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "hello jdk CompletableFuture," + str;
    }
}

Main Program ：

public class Main {
    public static void main(String[] args){
        CompletableFuture<String> future = CompletableFuture.supplyAsync(new MySupplier(" test "),
                Executors.newFixedThreadPool(1));
        future.whenCompleteAsync((r,t)->{
            System.out.println("(CompletableFuture) The result of asynchronous processing is :"+r);
        });
        System.out.println(" Deal with other logic 1");
        System.out.println(" Deal with other logic 2");
    }
}

Output results ：

 Deal with other logic 1
 Deal with other logic 2
(CompletableFuture) The result of asynchronous processing is :hello jdk CompletableFuture, test 

summary

1. stay JDK1.5 Provided in Future Pattern , Blocked when getting data , So other frameworks （guava and netty) Yes Future The pattern is extended , Callback functions are supported .
2. stay JDK1.8 Provided in CompletableFuture, Supports more powerful asynchronous callbacks 、 Streaming 、 Multiple Future The ability to combine processing .

Reference material

• Books Ge Yiming * 《Java High concurrency programming 》