Java multithreading — from Thread, Runnable, to CompletableFuture

1. Manually create multiple threads

1.1 the Thread and Runnable

Thread and Runnable are known to most readers, so skip them. Using Thread:

public class ThreadDemo {

	public static void main(String[] args) {
		Thread threadOne = new ThreadExample(Thread 1 "");
		Thread threadTwo = new ThreadExample(Thread 2 ""); threadOne.start(); threadTwo.start(); }}class ThreadExample extends Thread {

	private String name;

	public ThreadExample(String name) {
		this.name = name;
	}

	@Override
	public void run(a) {
		System.out.printf("Thread {%s} output \n", name); }}Copy the code

Using Runnable:

public class RunnableDemo {

	public static void main(String[] args) {
		Thread threadOne = new Thread(new RunnableExample(Thread 1 ""));
		Thread threadTwo = new Thread(new RunnableExample(Thread 2 "")); threadOne.start(); threadTwo.start(); }}class RunnableExample implements Runnable {

	private String name;

	public RunnableExample(String name) {
		this.name = name;
	}

	@Override
	public void run(a) {
		System.out.printf("Thread {%s} outputs %s", name, System.lineSeparator()); }}Copy the code

An implementation interface and an inherited class are not much different when used directly. Some places say that a Runnable object can be reused by multiple threads, but in fact it can only share some parameter Settings. What is actually useful is the Thread pool below, which only accepts arguments of type Runnable (or Callable), not of type Thread.

1.2 Callable and Future and FutureTask

For child threads, we might have two requirements:

1. Gets the result of the child thread running
2. Get child thread health status (success, failure, exception)

Thread does not meet these two requirements. Runnable can fetch state but not result, hence Callable. Callable can be used with a Future to obtain the results of child thread execution.

public class FutureDemo {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		Runnable one = new RunnableExample(Thread 1 "");
		Callable two = new CallableExample(Thread 2 "");
		ThreadPool.pool.submit(one);
        // Here the get method blocksFuture future = ThreadPool.pool.submit(two); System.out.println(future.get()); ThreadPool.pool.shutdown(); }}class CallableExample<T> implements Callable<T> {

	private T name;

	public CallableExample(T name) {
		this.name = name;
	}

	@Override
	public T call(a) {
		System.out.printf("Thread {%s} outputs %s", name, System.lineSeparator());
		returnname; }}class ThreadPool {

	private ThreadPool(a) {}public static final ThreadPoolExecutor pool = new ThreadPoolExecutor(
			5.10.10,
			TimeUnit.SECONDS,
			new LinkedBlockingQueue<>(100),
			r -> new Thread(r),
			new ThreadPoolExecutor.DiscardOldestPolicy());
}
Copy the code

Future.get() is a blocking method. To avoid blocking, we put multiple futures into a List, iterate through it, and get after isDone() determines that the child thread has finished executing. The Future also has a FutureTask that accepts Callable as a parameter.

public class FutureTaskDemo {

	public static void main(String[] args) throws ExecutionException, InterruptedException {
		Callable one = new CallableExample(Thread 1 "");
		Callable two = new CallableExample(Thread 2 "");
		FutureTask<CallableExample> task = newFutureTask<>(two); ThreadPool.pool.submit(one); ThreadPool.pool.submit(task); System.out.println(task.get()); ThreadPool.pool.shutdown(); }}Copy the code

Useful when it is not convenient to get the return value of submit.

2. The thread pool

Thread creation and destruction consume resources, thread context switching also consumes resources, and does not control the number of threads easy OOM, so there is a thread pool. You can use the following four thread pools:

• Executors.newCachedThreadPool();
• Executors.newFixedThreadPool(3);
• Executors.newScheduledThreadPool(3);
• Executors.newSingleThreadExecutor();

You can also customize:

ThreadPoolExecutor THREAD_POOL = new ThreadPoolExecutor(2.2.60,
			TimeUnit.SECONDS, new ArrayBlockingQueue<>(1),
			new ThreadFactoryBuilder().setNameFormat("HomePageCardFeatureQueryThreadPool-%d").setDaemon(true).build(),
			new ThreadPoolExecutor.DiscardPolicy());
Copy the code

. Among them, the ThreadPoolExecutor DiscardPolicy () thread pool is reached the maximum number of threads and the queue is full after processing strategy, ThreadFactoryBuilder () is the engineering method of generation of thread, this is Google’s guava bag, We can also implement the ThreadFactory in the java.util.Concurrent package ourselves. A simple example:

	public class MyThreadFactory implements ThreadFactory {
		private int counter;
		private String name;
		private List<String> stats;

		public MyThreadFactory(String name) {
			counter = 1;
			this.name = name;
		}

		@Override
		public Thread newThread(Runnable runnable) {
			Thread t = new Thread(runnable, name + "-Thread_" + counter);
			counter++;
			returnt; }}Copy the code

3. Insert a CompletionService

CompletionService returns results in the order in which the threads are finished, without iterating through the Future list

public class CompletionServiceDemo {

	public static void main(String[] args) {
		try {
			int taskCount = 10;
			CompletionService<Integer> completionService = new ExecutorCompletionService<>(ThreadPool.pool);

			// Method 1: The future is returned when the task is submitted, and the queue is traversed in the order in which the task is submitted
// List
      
       > futureList = new ArrayList<>();
      
// for (int i = 0; i < taskCount; i++) {
// futureList.add(completionService.submit(new demo.CallableExample<>(i + 1)));
/ /}
// for (Future
      
        future : futureList) {
      
// // Each thread blocks here in sequence, waiting for the task to complete
// Integer result = future.get();
Printf (" thread gets result {%2d}\n", result); // system.out. printf(" thread gets result {%2d}\n", result);
/ /}

			Method 2. Use the internal blocking queue of CompletionService
			for (int i = 0; i < taskCount; i++) {
				completionService.submit(new CallableExample<>(i + 1));
			}
			for (int i = 0; i < taskCount; i++) {
				// CompletionService.take () is used to maintain the blocking queue internally
				Integer result = completionService.take().get();
				System.out.printf(Thread gets execution result {%2d}%s, result, System.lineSeparator()); }}catch (Exception e) {
			e.printStackTrace();
		} finally {
			// Close the thread poolThreadPool.pool.shutdown(); }}}Copy the code

4. Powerful CompletableFuture provided by JDK8

public class CompletableFutureDemo {

	public static void main(String[] args) {
		int taskCount = 10;
		CompletableFuture[] futures = new CompletableFuture[taskCount];
		List<Integer> res = new CopyOnWriteArrayList<>();
		for (int i = 0; i < taskCount; i++) {
			int count = i;
			futures[i] = CompletableFuture.supplyAsync(
					() -> {
						System.out.printf(Thread gets execution result {%2d}%s, count, System.lineSeparator());
						return count;
					},
					ThreadPool.pool).thenAccept(x -> res.add(x));
		}
		CompletableFuture.allOf(futures).orTimeout(1000, TimeUnit.SECONDS).join();
		res.stream().forEach(System.out::println);
		ThreadPool.pool.shutdown();
	}
Copy the code

What if there is no timeout for the CompletableFuture in JDk1.8? You can use the get() method instead, or you can add a timed thread.

public class CompletableFuture8Demo {
    public static void main(String[] args) throws Exception {
        ScheduledThreadPoolExecutor delayerThreadPool = new ScheduledThreadPoolExecutor(1.new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return newThread(r); }});// Throw a timed exception with a timed task
        CompletableFuture<Integer> timeoutFuture = new CompletableFuture<>();
        delayerThreadPool.schedule(() -> {
            timeoutFuture.completeExceptionally(new TimeoutException("Time's up"));
        }, 1, TimeUnit.SECONDS);

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(
                () -> {
                    try {
                        Thread.sleep(1000);
                        System.out.println("1s");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        throw new RuntimeException(e.getMessage());
                    }
                    return 100;
                })
                .exceptionally(e -> 50);
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(
                () -> {
                    try {
                        Thread.sleep(2000);
                        System.out.println("2s");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        throw new RuntimeException(e.getMessage());
                    }
                    return 100;
                })
                .exceptionally(e -> 50);
        CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(
                () -> {
                    try {
                        Thread.sleep(3000);
                        System.out.println("3s");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        throw new RuntimeException(e.getMessage());
                    }
                    return 100;
                })
                .exceptionally(e -> 50);
        CompletableFuture<Integer> res = future
                .thenCombineAsync(future2, (i, j) -> i + j)
                .thenCombineAsync(future3, (i, j) -> i + j)
                .applyToEither(timeoutFuture, Function.identity());
        try {
            System.out.println(res.get());
        } catch (Exception e) {
            e.printStackTrace();
        } finally{ delayerThreadPool.shutdownNow(); }}}Copy the code