Future ＆ CompletionService

How to create threads

The ways to create threads are generally as follows  4  Kind of ：

• Inherit  Thread  class

• Realization  Runable  Interface

• Realization  Callable  Interface

• Using thread pools

among , Direct inheritance  Thread  Or realize  Runnable  Interfaces can create threads , But one problem with both methods is ： no return value , That is, you cannot get the result of execution . therefore  java1.5  Provided.  Callable  Interface to implement this scenario , coordination  Future  and  FutureTask  Use .

Why  Callable？Runnable  The defects of the system are as follows ：

• Cannot return a return value

• Can't throw  checked Exception

Callable  Of  call  Methods can have return values , You can declare to throw an exception . and  Callable  With one  Future  class , adopt  Future  You can understand the task execution , Or cancel the execution of the task , You can also get the results of task execution , These functions are  Runnable  Impossible .

public class FutureDemo {

 public static void main(String[] args) throws ExecutionException, InterruptedException {

 new Thread(() -> {
 System.out.println(" adopt  Runnable  Way to perform tasks ");
 }).start();


 FutureTask task = new FutureTask(new Callable() {
 @Override
 public Object call() throws Exception {
 System.out.println(" adopt  Callable  Way to perform tasks ");
 //  wait for  3s  Simulation execution task
 Thread.sleep(3000);
 return " Return to task result ";
 }
 });

 new Thread(task).start();

 System.out.println(" result ：" + task.get());
 }
}


Future

Future  Indicates a that may not have been completed

asynchronous

Results of the task , For specific  Runnable  perhaps  Callable  The execution result of the task is cancelled 、 Query complete 、 To get the results .

Yes, if necessary

get

Method to get the execution result , This method will

Blocking

Until the task returns the result .

Construction method

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


actually Callable = Runnable + result

.

Common methods

Future  The interface has the following methods ：

• boolean cancel (boolean mayInterruptIfRunning)

• Cancel the execution of the task . Parameter specifies whether to interrupt task execution immediately , Or wait until the mission is over .

• boolean isCancelled ()

• Whether the task has been cancelled , Cancel the task before it is completed , Then return to  true.

• boolean isDone ()

• Whether the task has been completed . It should be noted that if the task terminates normally 、 Exception or cancel , Will return to  true.

• V get () throws InterruptedException, ExecutionException

• Wait for the task to finish , Then get  V  Result of type .InterruptedException  Thread interrupted exception ,ExecutionException Task execution exception , If the mission is cancelled , And throw  CancellationException .

• V get(long timeout,TimeUnit unit) throws InterruptedException,ExecutionException, TimeoutException

• Same as above  get  Function as , Too much time out . Parameters  timeout  Specify the timeout period ,uint  Specify the unit of time , In enumerating classes TimeUnit There are related definitions in . If the calculation times out , Will throw out  TimeoutException.

FutureTask

Future  In practice  FutureTask  Realization , This object acts as a bridge between consumers and producers , Consumers go through  FutureTask  Store the processing results of the task , Update the status of the task ： Not yet begun 、 Dealing with 、 Finished waiting . And the producer gets  FutureTask  Be transformed into  Future  Interface , You can get the processing results of the task in a blocking way , Non blocking access to task processing status .

FutureTask  It can be regarded as  Runnable  To execute , It can also be regarded as  Future  To get  Callable  Return result of . The inheritance relationship is as follows ：

Usage mode

You can put  Callable  Example as  FutureTask  Arguments to the constructor , Generate  FutureTask  The object of , Then treat this object as a  Runnable  object , Put it into the thread pool or start another thread to execute , In the end, you can go through  FutureTask  Get the result of task execution .

public class FutureTaskDemo {

 public static void main(String[] args) throws ExecutionException, InterruptedException {
 //  Realization  Callable
 Task task = new Task();
 // structure futureTask
 FutureTask<Integer> futureTask = new FutureTask<>(task);
 // As Runnable Enter the reference
 new Thread(futureTask).start();

 System.out.println(&quot;task Running results ：&quot; + futureTask.get());
 }

 static class Task implements Callable<Integer> {

 @Override
 public Integer call() throws Exception {
 System.out.println(&quot; The child thread is calculating &quot;);
 int sum = 0;
 for (int i = 0; i < 100; i++) {
 sum += i;
 }
 return sum;
 }
 }
}


Use scenarios

A  The service call  B  Service multiple interfaces with the shortest response time ？

For example, product information query in promotional activities （ Including basic information of goods 、 commodity price 、 inventory 、 Commodity images 、 Commodity sales status, etc ）, Another example is the ticket order details page （ Including order information 、 Flight information 、 Passenger information 、 Contact information 、 Payment information 、 Information on the provisions for returning and changing signatures, etc , Usually, this information will not be provided by a service ）.

This information is distributed in different business centers , Services provided by different systems . If synchronous mode is adopted , Suppose an interface needs  50ms, The total time is the sum of the time spent on each interface , Then you need to query the information on an order details page  200ms-300ms, This is not satisfactory for us .

therefore , We can go through  Future  Get information asynchronously in parallel , as follows ：

public class FutureTaskDemo2 {

 public static void main(String[] args) throws ExecutionException, InterruptedException {

 FutureTask<String> ft1 = new FutureTask<>(new T1Task());
 FutureTask<String> ft2 = new FutureTask<>(new T2Task());
 FutureTask<String> ft3 = new FutureTask<>(new T3Task());
 FutureTask<String> ft4 = new FutureTask<>(new T4Task());
 FutureTask<String> ft5 = new FutureTask<>(new T5Task());

 // Build thread pool
 ExecutorService executorService = Executors.newFixedThreadPool(5);
 executorService.submit(ft1);
 executorService.submit(ft2);
 executorService.submit(ft3);
 executorService.submit(ft4);
 executorService.submit(ft5);
 // Get execution results
 System.out.println(ft1.get());
 System.out.println(ft2.get());
 System.out.println(ft3.get());
 System.out.println(ft4.get());
 System.out.println(ft5.get());

 executorService.shutdown();

 }

 static class T1Task implements Callable<String> {
 @Override
 public String call() throws Exception {
 System.out.println(&quot;T1: Query flight information ...&quot;);
 TimeUnit.MILLISECONDS.sleep(5000);
 return &quot; Flight information query succeeded &quot;;
 }
 }

 static class T2Task implements Callable<String> {
 @Override
 public String call() throws Exception {
 System.out.println(&quot;T2: Check the information of the passengers ...&quot;);
 TimeUnit.MILLISECONDS.sleep(50);
 return &quot; Flight attendant information query succeeded &quot;;
 }
 }

 static class T3Task implements Callable<String> {
 @Override
 public String call() throws Exception {
 System.out.println(&quot;T3: Query contact information ...&quot;);
 TimeUnit.MILLISECONDS.sleep(50);
 return &quot; Contact information query succeeded &quot;;
 }
 }

 static class T4Task implements Callable<String> {
 @Override
 public String call() throws Exception {
 System.out.println(&quot;T4: Query the refund and change information ...&quot;);
 TimeUnit.MILLISECONDS.sleep(50);
 return &quot; Refund and change information query succeeded &quot;;
 }
 }

 static class T5Task implements Callable<String> {
 @Override
 public String call() throws Exception {
 System.out.println(&quot;T5: Query other information ...&quot;);
 TimeUnit.MILLISECONDS.sleep(50);
 return &quot; Other information query succeeded &quot;;
 }
 }
}


Execution results ：

Future The limitations of

essentially ,

Future Represents the result of an asynchronous calculation

. It provides isDone() To check whether the calculation has been completed , And after the calculation , Can pass

get()

Method to get the calculation results . In asynchronous computing ,Future  It's really an excellent interface . however , It does have many limitations .

• Multitasking concurrently

• Future  Provided only  get()  Method to get the result , And it's blocked . therefore , There is no other way but to wait for you .

• Cannot chain call multiple tasks

• If you want to perform a specific action after the calculation task is completed , E-mail , but  Future  But it doesn't provide such ability .

• Cannot combine multiple tasks

• If you run  10  A mission , And expect to perform specific actions after they are all executed , So in  Future  There's nothing I can do about it .

• No exception handling

• Future  There are no exception handling methods in the interface .

CompletionService

Callable+Future  Can achieve more than one  task  Parallel execution , But if you encounter the previous  task  Slow execution requires

Block waiting

Ahead  task  After execution  task  To achieve results . and  CompletionService  Its main function is to generate tasks on one side , While getting the return value of the task . Separate the two things , Tasks don't block each other , You can get the first result after the first execution , No longer rely on task order .

Construction method

CompletionService  It's an interface , In fact, the implementation class is  ExecutorCompletionService,ExecutorCompletionService  Give the specific calculation task to  Executor  complete .

ExecutorCompletionService(Executor executor)

ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)


Common methods

• submit

• To submit a  Callable  perhaps  Runnable  Type of task , And back to  Future.

• take

• Blocking method , Get and remove the result of a completed task from the result queue , If not, it will block , Until a task is completed, the result is returned .

• poll

• Get and remove the result of a completed task from the result queue , If not, it will return  null, This method doesn't block .

Use scenarios

Inquiry from different e-commerce platforms , And keep the price .

During our inquiry , The return time of different e-commerce platforms must be different due to different factors , So for those that return first, let's save .

public class CompletionServiceDemo {

 public static void main(String[] args) throws InterruptedException, ExecutionException {

 // Creating a thread pool
 ExecutorService executor = Executors.newFixedThreadPool(10);
 // establish CompletionService
 CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
 // Asynchronous to e-commerce S1 inquiry
 cs.submit(() -> getPriceByS1());
 // Asynchronous to e-commerce S2 inquiry
 cs.submit(() -> getPriceByS2());
 // Asynchronous to e-commerce S3 inquiry
 cs.submit(() -> getPriceByS3());
 // Asynchronously save the inquiry result to the database
 for (int i = 0; i < 3; i++) {
 // Get from blocking queue futureTask
 Integer r = cs.take().get();
 executor.execute(() -> save(r));
 }
 executor.shutdown();
 }

 private static void save(Integer r) {
 System.out.println(&quot; Save inquiry results :&quot; + r);
 }

 private static Integer getPriceByS1() throws InterruptedException {
 TimeUnit.MILLISECONDS.sleep(5000);
 System.out.println(&quot; Online retailers S1 Inquiry information 1200&quot;);
 return 1200;
 }

 private static Integer getPriceByS2() throws InterruptedException {
 TimeUnit.MILLISECONDS.sleep(8000);
 System.out.println(&quot; Online retailers S2 Inquiry information 1000&quot;);
 return 1000;
 }

 private static Integer getPriceByS3() throws InterruptedException {
 TimeUnit.MILLISECONDS.sleep(3000);
 System.out.println(&quot; Online retailers S3 Inquiry information 800&quot;);
 return 800;
 }
}


Execution results ：

Realization principle

Through internal

Blocking queues +FutureTask

, The task is completed first, and the priority can be obtained , That is, the results are sorted according to the order of completion , There's a

First in first out blocking queue

, Used to save completed execution of  Future, By calling its  take  Method or  poll  Method can get a completed execution of  Future, And then by calling  Future  Interface implementation class  get  Method to get the final result .

On the implementation ,ExecutorCompletionService  A... Is created in the constructor  BlockingQueue（ Unbounded queue based on linked list used LinkedBlockingQueue）, The  BlockingQueue  Its purpose is to preserve  Executor  Results of execution . When the calculation is complete , call  FutureTask  Of  done  Method . When a task is submitted to  ExecutorCompletionService  when , First, wrap the task into  

QueueingFuture

, It is  FutureTask  A subclass of , And then rewrite  FutureTask  Of  done  Method , After the  Executor  The result of the calculation is put into  BlockingQueue  in .

When  ExecutorCompletionService  When you submit a task , as follows ：

Then package the task as  

QueueingFuture

,QueueingFuture  The source code is as follows ：

private static class QueueingFuture<V> extends FutureTask<Void> {
 QueueingFuture(RunnableFuture<V> task,
 BlockingQueue<Future<V>> completionQueue) {
 super(task, null);
 this.task = task;
 this.completionQueue = completionQueue;
 }
 private final Future<V> task;
 private final BlockingQueue<Future<V>> completionQueue;
 //  rewrite  FutureTask  Of  done  Method , After the  Executor  The result of the calculation is put into  BlockingQueue  in
 protected void done() { 
 completionQueue.add(task); 
 }
}


You can see that from the code ,CompletionService  Convert submitted tasks to  QueueingFuture, also

covers done Method

, stay  done  The method is to add the task to the task queue .

ExecutorService and CompletionService

stay  Future  We use thread pool  ExecutorService  The asynchronous way is used to execute tasks in parallel , But the problem is that when a task is submitted to the thread pool , Get a batch from the thread pool  

Future

  A collection of objects , then

Traverse in turn

Call its  

get()

  Method , and  

get()

  The method is blocked again , So if a  

Future

  Object execution time is too long , Due to the blockage, it is impossible to finish the work in time  

Future

  The results are .

CompletionService  The reason why we can do

While generating tasks , While getting the return value of the task

, Just because it doesn't take sequential traversal  

Future

  The way , Instead, a result queue is added in the middle , Put the results into the queue immediately after the task is completed , So what you get from the queue is the result of the earliest completion .

Performance optimization practice

There is a scenario , An examination system needs to extract some questions from the question bank according to some rules and generate a large number of offline practice documents （PDF）, One copy  PDF  The document generation process is as follows ：

• Isolate the problems that need to be solved （60~120  individual , The average is about  80  About two topics ）;

• Analyze and deal with the problem , Download the pictures in the title to the local , Then call the third party tool generation.  PDF  file （ It takes about  3~10  second ）;

• take  PDF  Upload document to  OSS  Cloud space for storage （ It takes about  1~3  second ）;

• Provide the document address for users to download and print .

According to the above rules, a  PDF  Document time consuming  

4~13s

Between , Suppose its average value is

8s

.

So generation  100  A document is  800s（ about  13  minute ）,1000  A document is  8000s（ about  133  minute ,2.2 Hours ）, Add up in turn .

In our first version , In order to launch the project , Direct serial mode , In short, it's a  

for

Loop to operate .

Simulate the above operation as follows ：

/**
 *  Pending document information as needed
 */
public class PDFDocVO {

 // Name of document to be processed
 private String docName;
 
 //  Omit others


 public String getDocName() {
 return docName;
 }

 public void setDocName(String docName) {
 this.docName = docName;
 }
}


Generate local actual document ：

public class ProduceDocService {

 /**
 *  Process the pending document as a local actual document
 *
 * @param doc
 * @return
 */
 public static String makePDF(PDFDocVO doc) throws InterruptedException {
 //  use  sleep  Time consuming range of simulated document generation  3~10s
 Random r = new Random();
 Thread.sleep(3000 + r.nextInt(7000));
 return "local" + doc.getDocName();
 }
}


Upload files ：

public class UploadDocService {

 /**
 *  Analog upload
 *
 * @param localName  Where the actual document is stored locally
 * @return oss  File path
 * @throws InterruptedException
 */
 public static String upload(String localName) throws InterruptedException {
 //  use  sleep  Time consuming range of simulated document generation  1~3s
 Random r = new Random();
 Thread.sleep(1000 + r.nextInt(2000));
 return "https://aliyun.oss.xxx/file/" + localName;
 }
}


Serial

import java.util.ArrayList;
import java.util.List;

public class SerializeModel {

 /**
 *  Eligible documents
 *
 * @param count
 * @return
 */
 public static List<PDFDocVO> getPDFDocList(int count) {
 List<PDFDocVO> list = new ArrayList<>();
 for (int i = 0; i < count; i++) {
 PDFDocVO doc1 = new PDFDocVO();
 list.add(doc1);
 }
 return list;
 }

 public static void main(String[] args) throws InterruptedException {
 List<PDFDocVO> docList = getPDFDocList(10);
 //  Starting time
 long start = System.currentTimeMillis();
 for (PDFDocVO doc : docList) {
 //  Generating documentation
 String localName = ProduceDocService.makePDF(doc);
 //  Upload documents
 UploadDocService.upload(localName);
 }
 long total = System.currentTimeMillis() - start;
 System.out.println(&quot; The total time is ：&quot; + total / 1000 + &quot; second ,&quot; + total / 1000 / 60 + &quot; minute &quot;);
 }
 
}


The test results are as follows ：

test  10  Tasks , The total time is  82s, That is to generate a  PDF  The average time taken for a document is  

82/10=8.2s

.

parallel

For parallel mode , By facing up to  Future  Understanding （ namely  Future  The limitations of  ）, For the above scene , Generating documents and uploading documents are related , If you want to generate documents and upload documents in parallel , When you upload a document, you don't know which local file it is , So just through  Future  Can't complete parallel tasks , So we can consider using  CompletionService, parallel + Asynchronous way .

parallel + asynchronous

For this offline document , The generation independence of each document is very high , Therefore, it is naturally suitable for multi-threaded concurrent processing , You can use the producer consumer model .

So when a caller's request is received , Package the data parsed by the request into a blocking queue , Then there will be multiple threads for consumption processing , That is to generate each specific document .

When the document is generated , Use the producer consumer model again , Put into another blocking queue , Another group of threads is responsible for uploading . When the upload is completed successfully , The upload thread returns the download address of the document , Indicates that the current document has completed successfully .

notes

： The thread pool cannot be directly used to submit tasks to the blocking queue , Once a task takes too long , Then the later task must be blocked until the previous task is completed , So we should use  CompletionService, That is, once a task generates a document successfully , Get and execute the upload first .

Code implementation ：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * TODO
 *
 * @date 2022/4/7 22:08
 */
public class ParallelAsyncModel {

 //  Thread pool threads
 public final static int THREAD_COUNT = Runtime.getRuntime().availableProcessors();

 //  Thread pool for processing document generation  IO Intensive task
 private static ExecutorService makeDocService = Executors.newFixedThreadPool(THREAD_COUNT * 2);

 //  Thread pool for processing document upload
 private static ExecutorService uploadDocService = Executors.newFixedThreadPool(THREAD_COUNT * 2);

 //  Document generation queue
 private static CompletionService<String> makeDocCompletionService = new ExecutorCompletionService(makeDocService);

 //  Document upload queue
 private static CompletionService<String> uploadDocCompletionService = new ExecutorCompletionService(uploadDocService);


 /**
 *  Eligible documents
 *
 * @param count
 * @return
 */
 public static List<PDFDocVO> getPDFDocList(int count) {
 List<PDFDocVO> list = new ArrayList<>();
 for (int i = 0; i < count; i++) {
 PDFDocVO doc1 = new PDFDocVO();
 list.add(doc1);
 }
 return list;
 }

 public static void main(String[] args) throws InterruptedException {
 int count = 100;
 List<PDFDocVO> docList = getPDFDocList(count);
 //  Starting time
 long start = System.currentTimeMillis();
 //  Multithreading document generation
 for (PDFDocVO doc : docList) {
 makeDocCompletionService.submit(new Callable<String>() {
 @Override
 public String call() throws Exception {
 String localName = ProduceDocService.makePDF(doc);
 return localName;
 }
 });
 }

 //  Upload documents
 for (int i = 0; i < count; i++) {
 Future<String> take = makeDocCompletionService.take();
 uploadDocCompletionService.submit(new Callable<String>() {
 @Override
 public String call() throws Exception {
 String uploadUrl = UploadDocService.upload(take.get());
 return uploadUrl;
 }
 });
 }
 long total = System.currentTimeMillis() - start;
 System.out.println(&quot; The total time is ：&quot; + total / 1000 + &quot; second ,&quot; + total / 1000 / 60 + &quot; minute &quot;);
 }
}


Let's adjust the number of tasks to  100, The test results are as follows ：

test  100  Tasks , The total time is  32s, That is to generate a  PDF  The average time taken for a document is  

32/100=0.32s

.

Compared with serial execution

8.2/0.32 ≈ 25  times

.

Thread number setting

For the setting of thread pool parameters, we usually follow the following principles ：

• IO  Intensive task , The number of threads is  
  CPU  The core number *2
  ;

• CPU  Intensive task , The number of threads is  
  CPU  The core number +1
  .

If we follow our general  IO  Intensive task , The two line process pools are set up by the machine  

CPU  The core number *2

, But is this the best ？

actually , Through repeated experiments, it is found that , The number of threads in the thread pool that process documents is set to  

CPU  The core number *4

, Continuing to increase the number of threads does not bring performance improvement .

Because the time of processing documents and uploading documents after improvement is basically  

1:4

  To  

1:3

  The appearance of , Therefore, the number of threads in the thread pool for processing documents is set to  

CPU  The core number *4*3

.

You can see , From the original  

32s  Reduced to  19s

.

therefore , For the number of threads in the thread pool , Too few threads will reduce the overall performance of the program , Too many threads will also consume memory and other resources , So if you want to be more accurate , Pressure measurement can be carried out , monitor  JVM  Thread situation and  CPU  The load of , Measure the number of threads that should be created according to the actual situation , Make rational and full use of resources .

summary

about Futrue  And  Callable

You can see that the asynchronous implementation with returned results ultimately depends on  FutureTask, It also realizes  Runnable  And  Future, To have a  Callable  attribute .

get  Methods according to the  FutureTask  The state of will suspend the thread and put it into the waiting list

了 .

At the same time, it can be used in the thread pool to be executed , It will be called at the end of the thread pool  run  Method ,

run  Method will call  Callable  Of  call  Method is the real calculation method , The result will be modified after it is returned  FutureTask  And wake up the thread waiting for the linked list

.

Thread pool  submit  Method also supports  Runnable  Parameters , however  FutureTask  Execution is  Callable  Of  call  Method , that  Runnable  Medium run  How to translate it into  Callable  Of  call  Methods? ？

It's actually quite simple ,

The adapter mode is adopted

, To build a  Callable  Subclasses of  RunnableAdapter, It saves a  Runnable  attribute ,RunnableAdapter  Of  call  Method call  Runnable  Of  run  Method , The returned result is  null  Or send a result yourself .

So we need to implement asynchronous tasks that return results , Or to achieve  Callable  And implement  call  Method , Or create a  Runnable  The implementation class is also OK .

about CompletionServic

CompletionServic  The submission task uses the submission of the thread pool , And he only created one himself FutureTask Subclasses of  QueueingFuture  Used to implement  done  Method , After the task is completed, put  FutureTask  Add to blocking queue .