[JUC learning road day 9] barrier derivatives

Semaphore

In most application environments , Many resources are actually Limited , for example ： The server provides 4 nucleus 8 Thread CPU resources , In this way, all applications no matter how preemptive , Finally, the only thing that can be preempted is this fixed 8 Threads to process the application , In fact, this is a limited resource . In the application environment facing large-scale concurrent access , In order to reasonably arrange the scheduling of limited resources , stay JUC Provided in Semaphore（ Semaphore ） Processing class .

Example ： Use semaphores to control the calling system

package juc.semaphore;

import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/** * @author birdy * @date 2022/6/25 2:53 PM */
public class Main {
    
    public static void main(String[] args) {
    
        Semaphore semaphore = new Semaphore(2);
        for (int i = 0; i < 10; i ++) {
    
            new Thread(() -> {
    
                try {
    
                    semaphore.acquire();
                    System.out.printf("【%s】 Start business ,  Current number of people waiting ：%d\n", 
                            Thread.currentThread().getName(),
                            semaphore.getQueueLength());
                    int randTime = ThreadLocalRandom.current().nextInt(1000);
                    TimeUnit.MILLISECONDS.sleep(randTime);
                    System.err.printf("【%s】 Completion of handling business \n", 
                            Thread.currentThread().getName());
                    semaphore.release();
                }catch (Exception e) {
    
                    e.printStackTrace();
                }
            }, " user -" + i).start();
        }
    }
}

CountDownLanch

It is a thread management mechanism based on countdown synchronization , for example ： Three sub threads are created in the main thread , The main thread must continue to execute downward after all the three sub threads have been executed , Now it can be based on CountDownLatch Set the number of waiting threads , Every time a child thread finishes executing, it will count -1.

Example ： Wait for the thread to complete the counting operation

package juc.cdl;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/** * @author birdy * @date 2022/6/25 3:09 PM */
public class Main {
    
    public static final int TOTAL_VISITOR = 10;
    public static void main(String[] args) {
    
        System.out.println(" The sightseeing tour group officially set out ~");
        CountDownLatch countDownLatch = new CountDownLatch(TOTAL_VISITOR);
        for (int i = 0; i < TOTAL_VISITOR; i ++) {
    
            new Thread(() -> {
    
                int visitTime = ThreadLocalRandom.current().nextInt(500);
                try {
    
                    TimeUnit.MILLISECONDS.sleep(visitTime);
                } catch (Exception e) {
    
                    e.printStackTrace();
                }finally {
    
                    countDownLatch.countDown();
                    System.out.printf("【%s】 I've come back ~  There is still left %d A good brother \n",
                            Thread.currentThread().getName(), countDownLatch.getCount());
                }
            }, "VISITOR - " + i).start();
        }
        try {
    
            boolean isComplete = countDownLatch.await(400, TimeUnit.MILLISECONDS);
            if (isComplete) {
    
                System.out.print(" Everyone ！");
            } else {
    
                System.err.printf(" And then there were %d position , Out of date ！\n", countDownLatch.getCount());
            }
        } catch (Exception e) {
    
            e.printStackTrace();
        } finally {
    
            System.out.println(" The bus departs ！");
        }
    }
}

CyclicBarrier

Barrier It means a fence , and Cyclic It represents a circular concept . When the waiting thread reaches param After , You can perform the corresponding Rannable process . But waiting for the thread to timeout will break the barrier , Therefore, all processes in the waiting state will trigger BrokenBarrierException abnormal . It is used to wait for how many threads trigger the operation .

Example ： Match players

package juc.CyclicBarrier;

import java.util.concurrent.*;

/** * @author birdy * @date 2022/6/25 3:56 PM */
public class Main {
    
    public static void main(String[] args) {
    
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4, () -> {
    
            System.err.println(" The match is successful , take off ！");

        });
        for (int i = 0; i < 11; i ++) {
    
            new Thread(() -> {
    
                try {
    
                    int randTime = ThreadLocalRandom.current().nextInt(2000);
                    TimeUnit.MILLISECONDS.sleep(randTime);
                    System.out.printf(" A player has entered ：%s \t", Thread.currentThread().getName());
                    cyclicBarrier.await(500, TimeUnit.MILLISECONDS);
                } catch (TimeoutException e) {
    
                    System.err.println(" Player matching failed :"+ Thread.currentThread().getName());
                    cyclicBarrier.reset();
                } catch (BrokenBarrierException e) {
    
                    System.err.println(Thread.currentThread().getName() + " I can't wait for birds ！");
                }catch (Exception e) {
    
                    e.printStackTrace();
                }
            }, "user - " + i).start();
        }
    }
}

A player has entered ：user - 9 A player has entered ：user - 6 A player has entered ：user - 4 A player has entered ：user - 7 The match is successful , take off ！
A player has entered ：user - 2 A player has entered ：user - 5 A player has entered ：user - 10 user - 10 I can't wait for birds ！
Player matching failed :user - 2
user - 5 I can't wait for birds ！
A player has entered ：user - 3 A player has entered ：user - 0 A player has entered ：user - 8 A player has entered ：user - 1 The match is successful , take off ！

Exchanger

exchanger . A thread is calling exchanger.exchange() after , Will arrive exchange The switching point enters the waiting state , Until another thread exchanger Also exchange after arriving at the exchange point .

Example ： State exchange between producers and consumers

package juc.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/** * @author birdy * @date 2022/6/26 4:50 PM */
public class Main {
    
    public static final int TOTAL = 10;
    public static void main(String[] args) {
    
        Exchanger<String> exchanger = new Exchanger<>();
        new Thread(() -> {
    
            for (int i = 0; i < TOTAL; i ++) {
    
                try {
    
                    String msg = "hello - " + i;
                    int randInt = ThreadLocalRandom.current().nextInt(500);
                    TimeUnit.MILLISECONDS.sleep(randInt);
                    exchanger.exchange(msg);
                    System.out.println("【p】" + msg);
                } catch (InterruptedException e) {
    
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(() -> {
    
            for (int i = 0; i < TOTAL - 1; i ++) {
    
                try {
    
                    String exchange = exchanger.exchange(null);
                    System.err.println("【c】" + exchange);
                } catch (InterruptedException e) {
    
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

CompletableFuture

stay JDK1.5 Then a new multi-threaded processing interface is provided ：Callable, This interface needs to be connected with Future Interfaces are integrated , And then make the final asynchronous operation call , It improves the processing performance of multithreading .
JDK 1.5 Provided Future It can realize asynchronous computing operations , although Future The related methods of provide the execution ability of asynchronous tasks , However, the acquisition of thread execution results can only be processed by blocking or polling , The blocking method is different from the original intention of multi-threaded asynchronous processing , Polling will cause CPU Waste of resources , At the same time, we can't get the results in time . In order to solve these design problems from JDK1.8 Began to offer Future The extension implementation class of CompletableFuture, It can help developers simplify the complexity of asynchronous programming , At the same time, it can be combined with the functional programming mode to use the callback method to carry out asynchronous processing and calculation operations .

Example ： Simulate taxi system

package juc.completable_future;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/** * @author birdy * @date 2022/6/27 9:14 PM */
public class Main {
    
    public static final int DRIVER_NUMBER = 6;
    public static void main(String[] args) {
    
        CompletableFuture<String> future = new CompletableFuture<>();
        for(int i = 0; i < DRIVER_NUMBER; i ++) {
    
            new Thread(() -> {
    
                try {
    
                    System.out.printf("【%s】 Start taking orders ！\n", Thread.currentThread().getName());
                    String passenger = future.get();
                    System.out.printf("【%s】 Receive a new order ：%s\n", Thread.currentThread().getName(), passenger);
                } catch (Exception e) {
    
                    e.printStackTrace();
                }
            }, " Dripping car owner  - " + i).start();
        }
        new Thread(() -> {
    
            try {
    
                TimeUnit.MILLISECONDS.sleep(1000);
                future.complete(Thread.currentThread().getName());
            } catch (Exception e) {
    
                e.printStackTrace();
            }
        }, " Passengers to Tencent building ").start();
    }
}

【 Dripping car owner - 3】 Start taking orders ！
【 Dripping car owner - 2】 Start taking orders ！
【 Dripping car owner - 0】 Start taking orders ！
【 Dripping car owner - 0】 Receive a new order ： Passengers to Tencent building
【 Dripping car owner - 5】 Receive a new order ： Passengers to Tencent building
【 Dripping car owner - 2】 Receive a new order ： Passengers to Tencent building
【 Dripping car owner - 1】 Receive a new order ： Passengers to Tencent building

The above processing is based on synchronization , Except through complete() Method to remove the blocking state of all threads CompletableFuture You can also go through runAsync() Method defines an asynchronous task processing thread （ adopt Runnable Interface implementation ）, And the blocking state of all sub threads will not be released until the execution of the thread is completed .

Example ： Use runAsync() Asynchronous call

package juc.completable_future;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/** * @author birdy * @date 2022/6/27 9:24 PM */
public class Main {
    
    public static final int DRIVER_NUMBER = 6;
    public static void main(String[] args) {
    
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    
            try {
    
                TimeUnit.MILLISECONDS.sleep(5000);
                System.err.println(" There are no passengers nearby , Please try again later ！");
            } catch (InterruptedException e) {
    
                e.printStackTrace();
            }
        });
        for(int i = 0; i < DRIVER_NUMBER; i ++) {
    
            new Thread(() -> {
    
                try {
    
                    System.out.printf("【%s】 Start taking orders ！\n", Thread.currentThread().getName());
                    future.get();  //  Using asynchronous calls does not return a value 
                } catch (Exception e) {
    
                    e.printStackTrace();
                }
            }, " Dripping car owner  - " + i).start();
        }
    }
}

【 Dripping car owner - 2】 Start taking orders ！
【 Dripping car owner - 3】 Start taking orders ！
【 Dripping car owner - 4】 Start taking orders ！
【 Dripping car owner - 5】 Start taking orders ！
There are no passengers nearby , Please try again later ！