Java并发编程常用API应用场景总结_并发编程api

1 ReentrantLock与Condition

ReentrantLock是Java中的一个锁实现，它提供了与synchronized关键字类似的线程互斥机制，但具有一些额外的功能。它允许线程在锁定资源时以可重入的方式调用同步方法，并且可以灵活地控制锁的获取和释放。

与ReentrantLock一起使用的一个常见的功能是Condition，它可以用来实现更复杂的线程协调。Condition可以让线程在等待某些条件满足时进入休眠状态，并且在条件满足时唤醒它们。这使得线程能够更灵活地协调和通信。

总的来说，ReentrantLock和Condition提供了比synchronized关键字更灵活和强大的线程同步和协调机制，使得多线程编程更加容易和高效。

ReentrantLock与Condition在应用开发中有许多常见的应用场景。其中一个常见的应用场景是生产者-消费者模型。假设有一个共享的缓冲区，生产者负责往缓冲区中放入数据，而消费者从中取出数据。在这种情况下，ReentrantLock和Condition可以用来实现对缓冲区的线程安全访问和线程之间的协调。

下面是一个简单的Java代码示例，演示了如何使用ReentrantLock和Condition来实现生产者-消费者模型：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumerExample {
    private ReentrantLock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private int buffer;

    public void produce() {
        lock.lock();
        try {
            while (buffer != 0) {
                condition.await();
            }
            buffer = 1;
            System.out.println("Produced: " + buffer);
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void consume() {
        lock.lock();
        try {
            while (buffer == 0) {
                condition.await();
            }
            buffer = 0;
            System.out.println("Consumed: " + buffer);
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

在这个示例中，ReentrantLock被用来实现对共享缓冲区的互斥访问，而Condition被用来在生产者和消费者之间进行信号的发送和接收，以实现线程之间的协调。生产者和消费者可调用produce()和consume()方法来进行生产和消费操作。

这个示例展示了ReentrantLock与Condition在生产者-消费者模型中的典型应用场景，它们也可以用于其他需要复杂线程协调的场景，如线程池、消息传递等。

2 Semaphore

Semaphore（信号量）是计算机科学中的一个概念，用于控制对共享资源的访问。它是一个计数器，用来限制同时访问共享资源的线程数量。Semaphore维护一个计数器，对于每次资源的请求，计数器会递减，当计数器为0时，资源将不可用，线程将被阻塞直到有其它线程释放资源增加了计数器，使得资源再次可用。

Semaphore常用于解决资源池管理、线程并发访问控制等问题。它提供了灵活的机制，可以根据需求允许多个线程同时访问共享资源，或者限制资源的并发访问数量。

在Java中，Semaphore的实现类是java.util.concurrent.Semaphore，它提供了acquire()和release()等方法来分别获取和释放信号量。Semaphore也可以被用来实现生产者-消费者模型、限流及并发访问控制等场景。

总的来说，Semaphore是一个重要的多线程协作工具，能够有效地管理并发访问共享资源，提高程序的可靠性和性能。

Semaphore在Java中有许多常见的应用场景，其中一些包括限流、资源池管理、并发访问控制等。下面是一个常见的应用场景及代码示例：

1. 限流

在一些场景下，我们希望限制并发访问某个资源的线程数量，以防止资源被过度消耗。Semaphore可以用来实现这样的限流机制。

import java.util.concurrent.Semaphore;

public class ResourcePool {
    private Semaphore semaphore;

    public ResourcePool(int permits) {
        semaphore = new Semaphore(permits);
    }

    public void useResource() {
        try {
            semaphore.acquire();
            // 访问共享资源的代码
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

在上面的示例中，ResourcePool类使用Semaphore来限制并发访问共享资源的线程数量。当有线程需要访问资源时，它首先要获取semaphore的一个许可。如果没有许可可用（即达到了限流的最大数量），线程会被阻塞直到有许可可用。

2. 资源池管理

Semaphore还可以用于管理有限的资源池，并控制对资源的访问。

import java.util.concurrent.Semaphore;

public class ConnectionPool {
    private boolean[] connections;
    private Semaphore semaphore;

    public ConnectionPool(int poolSize) {
        connections = new boolean[poolSize];
        semaphore = new Semaphore(poolSize, true);
    }

    public int getResource() {
        try {
            semaphore.acquire();
            return getAvailableConnection();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return -1;
    }

    public void releaseResource(int index) {
        if (markAsUnused(index)) {
            semaphore.release();
        }
    }

    private synchronized int getAvailableConnection() {
        for (int i = 0; i < connections.length; i++) {
            if (!connections[i]) {
                connections[i] = true;
                return i;
            }
        }
        return -1;
    }

    private synchronized boolean markAsUnused(int index) {
        if (index < 0 || index >= connections.length) {
            return false;
        }
        if (connections[index]) {
            connections[index] = false;
            return true;
        }
        return false;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

在这个示例中，ConnectionPool类使用Semaphore来管理一个有限的连接池。Semaphore的许可数对应着连接池中可用的连接数量。当有线程需要获取连接时，它首先要获取semaphore的一个许可。获取到许可之后，线程可以从连接池中获取连接，并在使用完之后释放连接，同时释放Semaphore的许可。

以上是两个常见的应用场景以及相应的代码示例，展示了Semaphore在Java应用开发中的实陵用。Semaphore还可以在许多其它并发控制的场景中发挥作用，如控制并发任务的执行、资源池管理等。

3 Exchanger

在Java中，Exchanger是一个用于两个线程之间交换数据的同步工具。它提供了一个同步点，允许两个线程在这个同步点上交换数据。

Exchanger中的每个线程将会调用exchange()方法，并等待另一个线程到达交换点。当两个线程都到达交换点时，它们交换的是数据。在交换完成后，它们可以继续执行。

Exchanger的主要特性包括：

• 可以在两个线程之间进行数据交换
• 可以用于实现两个线程之间的协作
• 可以避免线程间的竞争条件

以下是一个简单的示例，演示了Exchanger的基本用法：

import java.util.concurrent.Exchanger;

public class ExchangerExample {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();

        // 线程1
        new Thread(() -> {
            String data1 = "Hello from Thread 1";
            try {
                String data2 = exchanger.exchange(data1);
                System.out.println("Thread 1 received: " + data2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 线程2
        new Thread(() -> {
            String data2 = "Hello from Thread 2";
            try {
                String data1 = exchanger.exchange(data2);
                System.out.println("Thread 2 received: " + data1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

在这个示例中，当线程1和线程2分别调用exchange()方法时，它们会等待对方到达交换点。一旦两个线程都到达交换点，它们交换的是数据。在这个示例中，线程1将自己的字符串传递给线程2，并接收线程2传递的字符串。

总的来说，Exchanger是一个有用的工具，用于在两个线程之间进行数据交换，可以用于协作任务，避免竞争条件等场景。

4 CountDownLatch

在Java中，CountDownLatch是一个同步工具类，用于协调多个线程之间的同步。它允许一个或多个线程等待直到其他线程执行完一系列操作后再继续执行。CountDownLatch的主要思想是，一个线程等待其他线程完成一些操作后再去执行。

CountDownLatch内部维护了一个计数器，它初始化时设置一个初始计数值，当计数值减至0时，等待的线程将被唤醒。每一个线程在完成自己的操作后，将计数值减1，直到计数值为0。在这个时刻，所有等待的线程得以继续执行。

常见的应用场景包括：

1. 在主线程等待多个子线程都完成后再执行某些操作
2. 控制并发任务的执行顺序
3. 在多个线程开始执行之前，做一些初始化工作
4. 在多个线程执行完毕后进行汇总处理

以下是一个简单的示例代码，演示了CountDownLatch的基本用法：

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);

        Worker worker1 = new Worker("Worker 1", latch);
        Worker worker2 = new Worker("Worker 2", latch);
        Worker worker3 = new Worker("Worker 3", latch);

        worker1.start();
        worker2.start();
        worker3.start();

        latch.await(); // 主线程在这里等待，直到所有Worker线程完成任务

        System.out.println("All workers have finished, main thread can continue.");
    }
}

class Worker extends Thread {
    private CountDownLatch latch;

    public Worker(String name, CountDownLatch latch) {
        super(name);
        this.latch = latch;
    }

    public void run() {
        System.out.println(Thread.currentThread().getName() + " is working");
        // 模拟工作
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + " has finished work");
        latch.countDown(); // 每个Worker线程完成工作后调用countDown
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

在这个示例中，主线程创建了三个Worker线程，并将同一个CountDownLatch对象传递给它们。每个Worker线程在完成工作后会调用countDown()方法来将CountDownLatch的计数器减1。当全部的Worker线程都完成工作后，主线程调用await()来等待，直到CountDownLatch的计数器变为0。

总的来说，CountDownLatch是用于线程之间互相等待的一种同步工具，它在多线程协作的场景中非常有用。常见的应用包括等待多个线程都完成后再执行某些操作，或者控制多个线程的执行顺序等。

5 CyclicBarrier

在Java中，CyclicBarrier是一个同步工具类，它允许一组线程在到达某个屏障点之前相互等待，然后在屏障点到达之后，继续执行。CyclicBarrier的主要特点是可以重用，一旦所有线程都到达屏障点，CyclicBarrier会复位，允许下一轮的等待和执行。

CyclicBarrier内部也维护着一个计数器，并且包含一个屏障动作，当线程到达屏障点时可以执行这个屏障动作。当计数值为0时，屏障点会打开，并且等待中的线程可以继续执行。

常见的应用场景包括：

1. 在多个线程分阶段执行任务时，等待所有线程都完成某一阶段的任务后再继续下一阶段。

2. 控制多个任务同时开始执行，等到所有任务都准备好之后再一起开始。

3. 在游戏编程中，可以用作等待所有玩家都准备好，然后一起进入游戏场景。

   以下是一个简单的示例代码，演示了CyclicBarrier的基本用法：

   import java.util.concurrent.BrokenBarrierException;
   import java.util.concurrent.CyclicBarrier;
   
   public class CyclicBarrierExample {
       public static void main(String[] args) {
           CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
               @Override
               public void run() {
                   System.out.println("All players are ready, game starts!");
               }
           });
   
           Player player1 = new Player("Player 1", barrier);
           Player player2 = new Player("Player 2", barrier);
           Player player3 = new Player("Player 3", barrier);
   
           player1.start();
           player2.start();
           player3.start();
       }
   }
   
   class Player extends Thread {
       private CyclicBarrier barrier;
   
       public Player(String name, CyclicBarrier barrier) {
           super(name);
           this.barrier = barrier;
       }
   
       public void run() {
           System.out.println(Thread.currentThread().getName() + " is ready");
           try {
               barrier.await(); // 等待其他玩家都准备好
               System.out.println(Thread.currentThread().getName() + " starts playing");
           } catch (InterruptedException | BrokenBarrierException e) {
               e.printStackTrace();
           }
       }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40

   在这个示例中，主线程创建了一个CyclicBarrier对象，其中包含3个参与者。每个参与者(Player)在准备好后调用await()方法等待其他玩家都准备好。当所有玩家都准备好之后，屏障点会打开并执行CyclicBarrier的Runnable。所有等待的线程可以继续执行。这里模拟了多个玩家在游戏开始前等待对方都准备好的场景。

   总的来说，CyclicBarrier是用于线程之间协同工作的一种同步工具，它在多线程协作的场景中非常有用。常见的应用包括等待多个线程都准备好后再继续执行某些操作，或者控制多个任务同时开始执行等。

4. 多个请求合并请求结果。

   import java.util.concurrent.BrokenBarrierException;
   import java.util.concurrent.CyclicBarrier;
   
   public class RequestThread implements Runnable {
       private CyclicBarrier barrier;
       private String request;
       private ResultHolder resultHolder;
   
       public RequestThread(CyclicBarrier barrier, String request, ResultHolder resultHolder) {
           this.barrier = barrier;
           this.request = request;
           this.resultHolder = resultHolder;
       }
   
       @Override
       public void run() {
           // 发起请求
           String result = sendRequest(request);
   
           try {
               // 等待其他线程完成请求
               barrier.await();
           } catch (InterruptedException | BrokenBarrierException e) {
               e.printStackTrace();
           }
   
           // 合并请求结果
           resultHolder.addResult(result);
       }
   
       private String sendRequest(String request) {
           // 发送请求并返回结果
           // 这里只是示例，你需要根据你的具体业务逻辑实现
           return "Result for request: " + request;
       }
   }
   
   public class ResultHolder {
       private List<String> results = new ArrayList<>();
   
       public synchronized void addResult(String result) {
           results.add(result);
       }
   
       public List<String> getResults() {
           return results;
       }
   }
   
   public class Main {
       public static void main(String[] args) {
           int numThreads = 5; // 假设有5个线程
           CyclicBarrier barrier = new CyclicBarrier(numThreads, () -> {
               // 在所有线程都到达barrier时触发的回调方法
               // 这里可以进行结果的合并操作
               ResultHolder resultHolder = new ResultHolder();
               List<String> results = resultHolder.getResults();
               for (String result : results) {
                   System.out.println(result);
               }
           });
   
           // 创建并启动线程
           for (int i = 0; i < numThreads; i++) {
               RequestThread requestThread = new RequestThread(barrier, "Request " + i, resultHolder);
               Thread thread = new Thread(requestThread);
               thread.start();
           }
       }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66
   67
   68
   69
   70

6 Phaser

在Java中，Phaser是一个同步工具类，可以用于协调多个线程的执行，并且提供了更灵活的控制机制。与CycliBarrier相比，Phaser支持更多的灵活性和功能。它能够支持任意数量的注册线程，并且可以在多个阶段进行控制。Phaser还具有分阶段执行任务、优雅处理终止等功能。

Phaser有几个主要的概念：

• Phase（阶段）：Phaser可以分为多个阶段，每个阶段结束后，线程可以继续向下执行。
• Arrive（到达）：表示线程到达某个阶段。
• Register（注册）：表示线程注册到Phaser中。

常见的应用场景包括：

1. 在多个线程分阶段执行任务时，等待所有线程都完成某一阶段的任务后再继续下一阶段。
2. 在分布式系统中进行多个任务的协同工作。
3. 任务分解和合并。

以下是一个简单的示例代码，演示了Phaser的基本用法：

import java.util.concurrent.Phaser;

public class PhaserExample {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(3);

        MyTask task1 = new MyTask("Task 1", phaser);
        MyTask task2 = new MyTask("Task 2", phaser);
        MyTask task3 = new MyTask("Task 3", phaser);

        task1.start();
        task2.start();
        task3.start();

        // Do something else while tasks are executing....

        // Wait for all tasks to complete
        int phase = phaser.awaitAdvance(0);
        System.out.println("All tasks have finished phase " + phase);
    }
}

class MyTask extends Thread {
    private Phaser phaser;

    public MyTask(String name, Phaser phaser) {
        super(name);
        this.phaser = phaser;
    }

    public void run() {
        System.out.println(Thread.currentThread().getName() + " is doing something");
        phaser.arriveAndAwaitAdvance(); // 到达一个阶段并等待其他线程
        System.out.println(Thread.currentThread().getName() + " continues to do something else");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

在这个示例中，主线程创建了一个Phaser对象，其中包含3个参与者。每个参与者在执行任务时，会调用arriveAndAwaitAdvance()方法等待其他任务都完成。当所有任务都到达并等待后，Phaser会进入下一个阶段。主线程可以调用awaitAdvance()方法等待所有任务完成。

总的来说，Phaser是一个用于协调多个线程执行的灵活同步工具类，它在多线程协作的场景中非常有用。常见的应用包括等待多个线程都完成某一阶段的任务后再继续下一阶段，以及任务分解和合并等。

7 Executor

接口 Executor 仅仅是一种规范，是一种声明，是一种定义，并没有实现任何的功能，所以大多数的情况下，需要使用接口的实现类来完成指定的功能，比如 ThreadPoolExecutor 类就是 Executor 的实现类，但 ThreadPoolExecutor 在使用上并不是那么方便，在实例化时需要传入很多个参数，还要考虑线程的并发数等与线程池运行效率有关的参数，所以官方建议使用Executors 工厂类来创建线程池对象。

7.1 Executors工厂

7.1.1 Executors.newCachedThreadPool()

在Java中，Executors.newCachedThreadPool()是用于创建一个可缓存的线程池的工厂方法。可缓存的线程池实际上是一种具有自动线程回收功能的线程池，可以根据需要自动扩展线程池的大小。如果线程池中的线程在60秒内都没有被使用，那么这些线程将被终止并从线程池中移除。当任务数增加时，线程池增加线程，当任务数减少时，线程池缩减线程。

常见的应用场景包括：

1. 适合执行很多短期异步的小任务，可灵活回收空闲线程，适用于执行大量的耗时较短的任务。
2. 通常在执行大量的异步任务而不知道任务到底有多少时使用，比如服务器端处理短连接。

以下是一个常见应用示例：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();

        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Executing task " + taskNumber + " on thread " + Thread.currentThread().getName());
            });
        }

        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

在这个示例中，通过Executors.newCachedThreadPool()方法创建了一个可缓存线程池。然后通过循环提交了10个任务给线程池执行。由于是可缓存线程池，如果有任务提交时没有空闲线程可用，线程池会自动创建新线程来执行任务。当线程闲置时间超过60秒后，会被终止并从线程池中移除。

总的来说，Executors.newCachedThreadPool()适用于处理大量的短期异步的小任务，灵活地回收线程资源，适用于执行大量的耗时较短的任务。在业务开发中，它通常用于服务器端处理短连接或者执行大量的异步任务。

当使用 Executors.newCachedThreadPool() 方法创建一个可缓存的线程池时，会使用默认的线程工厂来创建新线程。如果需要自定义线程工厂来定制线程的创建过程，可以使用 ThreadFactory 接口来实现。下面是一个示例代码，演示了如何自定义线程工厂来创建可缓存的线程池：

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class CustomThreadFactoryExample {
    public static void main(String[] args) {
        ThreadFactory customThreadFactory = new CustomThreadFactory("CustomThread");
        ExecutorService executor = Executors.newCachedThreadPool(customThreadFactory);

        for (int i = 0; i < 5; i++) {
            int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Executing task " + taskNumber + " on thread " + Thread.currentThread().getName());
            });
        }

        executor.shutdown();
    }
}

class CustomThreadFactory implements ThreadFactory {
    private String namePrefix;

    public CustomThreadFactory(String namePrefix) {
        this.namePrefix = namePrefix;
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, namePrefix + "-Thread-" + (int)(Math.random() * 100));
        return t;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

在这个示例中，我们首先实现了一个自定义的线程工厂 CustomThreadFactory 实现了 ThreadFactory 接口，重写了 newThread 方法，用于创建新的线程。在线程工厂中，我们可以为新创建的线程设置名称、优先级、守护状态等属性。

然后，我们将自定义线程工厂传递到 Executors.newCachedThreadPool(customThreadFactory) 方法中，以创建一个可缓存的线程池。

在执行过程中，我们可以看到每个任务都是由自定义的线程工厂创建出来的线程来执行的。

通过自定义线程工厂，我们可以更灵活地管理线程的创建过程，根据需求定制线程的属性，并且能够更好地定位线程的问题和调试。

7.1.2 Executors.newFixedThreadPool(int)

Executors.newFixedThreadPool(int) 是 Java 中用于创建固定大小的线程池的工厂方法。它将创建一个固定大小的线程池，即线程池中的线程数量是固定的，不会动态增加或减少。当有新的任务提交时，如果线程池中的线程都在执行任务，新的任务将会被放入任务队列中等待执行。

常见的业务开发中的应用示例包括：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5); // 创建一个固定大小为5的线程池

        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Executing task " + taskNumber + " on thread " + Thread.currentThread().getName());
            });
        }

        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

在这个示例中，通过 Executors.newFixedThreadPool(5) 方法创建了一个固定大小为5的线程池。然后通过循环提交了10个任务给线程池执行。由于线程池大小为5，因此只能有5个任务同时执行，其余的任务会被放入任务队列中等待执行。

固定大小的线程池适用于需要限制并行执行任务数量的场景，比如控制并发访问某个资源或服务。在很多业务应用中，特别是在服务器端开发中，经常需要限制同时执行的任务数量，而固定大小的线程池正好可以满足这种需求。

它也支持自定义线程工厂，方式同上。

7.1.3 Executors.newSingleThreadExecutor()

Executors.newSingleThreadExecutor() 是 Java 中用于创建仅包含单个线程的线程池的工厂方法。它将创建一个只有一个工作线程的线程池，确保所有任务都在同一个线程中按顺序执行。

常见的业务开发中的应用示例包括：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SingleThreadExecutorExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor(); // 创建一个只有单个线程的线程池

        for (int i = 0; i < 5; i++) {
            int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Executing task " + taskNumber + " on thread " + Thread.currentThread().getName());
            });
        }

        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

在这个示例中，通过 Executors.newSingleThreadExecutor() 方法创建了一个只有单个线程的线程池。然后通过循环提交了5个任务给线程池执行。由于线程池中只有一个工作线程，因此所有任务都会按顺序在同一个线程中执行。这可以确保任务之间的顺序性，适用于一些需要按序处理任务的场景。

在业务开发中，newSingleThreadExecutor() 常用于需要保证顺序执行的任务，或者需要在一个独立的线程中执行任务以避免并发问题的情况。例如，定时任务调度、消息队列消费等场景通常会使用单线程池来保证任务的顺序执行或避免并发问题。

它也支持自定义线程工厂，方式同上。

7.2 ThreadPoolExecutor

类ThreadPoolExecutor 可以非常方便地创建线程池对象，而不需要程序员设计大量的new实例化Thread相关的代码。

使用Executors工厂类的newXXXThreadExecutor()方法可以快速方便地创建线程池，但创建的细节却未知，通过查看源代码在调用newSingleThreadExecutor0方法时内部其实是实例化了1个ThreadPoolExecutor类的实例，源代码如下:

public static ExecutorService newSingleThreadExecutor() (return newFinalizableDelegatedExecutorService(new ThreadPoolExecutor(1,1,0L,TimeUnitMILLISECONDS,new LinkedBlockingOueue<Runnable>()));
1

ThreadPoolExecutor 是 Java 中用于自定义线程池的类，它提供了丰富的配置选项，可以用来创建各种类型的线程池，如固定大小线程池、缓存线程池、定时线程池等。

ThreadPoolExecutor 的常见构造方法如下：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, 
                   BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
1
2

其中，各个参数的含义如下：

• corePoolSize：核心线程池大小，即线程池中始终保持存活的线程数量。
• maximumPoolSize：最大线程池大小，线程池中允许存在的最大线程数量。
• keepAliveTime：非核心线程的空闲线程存活时间。
• unit：空闲线程存活时间的时间单位。
• workQueue：线程池中的任务队列，用来存放等待执行的任务。
• threadFactory：线程工厂，用来创建新的线程。
• handler：任务拒绝策略，当任务无法被执行时的处理方式。

业务开发中的常见应用代码示例如下：

import java.util.concurrent.*;

public class CustomThreadPoolExecutorExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, workQueue, threadFactory, handler);

        for (int i = 0; i < 20; i++) {
            int taskNumber = i;
            executor.execute(() -> {
                System.out.println("Executing task " + taskNumber + " on thread " + Thread.currentThread().getName());
            });
        }

        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

在这个示例中，我们创建了一个自定义的 ThreadPoolExecutor，配置了核心线程池大小为5，最大线程池大小为10，空闲线程的存活时间为60秒，使用了一个队列容量为10的任务队列，并且使用了默认的线程工厂和拒绝策略。然后我们提交了20个任务给线程池执行。这个示例展示了如何使用 ThreadPoolExecutor 来创建自定义的线程池，并提交任务执行。

方法 shutdown() 的作用是使当前未执行完的线程继续执行，而不再添加新的任务 Task，还有 shutdown() 方法不会阻塞，调用 shutdown() 方法后，主线程 main 就马上结束了，而线程池会继续运行直到所有任务执行完才会停止。如果不调用 shutdown() 方法，那么线程池会一直保持下去，以便随时执行被添加的新 Task 任务。
方法 shutdownNow() 的作用是中断所有的任务 Task，并且抛出IterruptedException 异常，前提是在 Runnable 中使用 if (Thread.currentThread.isInterrupted()== true) 语句来判断当前线程的中断状态，而未执行的线程不再执行，也就是从执行队列中清除。如果没有 if(Thread.currentThread.isInterrupted() == true)语句及抛出异常的代码，则池中正在运行的线程直到执行完毕，而未执行的线程不再执行，也从执行队列中清除。

当线程池调用 shutdown()方法时，线程池的状态则立刻变成 SHUTDOWN状态，此时不能再往线程池中添加任何任务，否则将会抛出 RejectedExecutionException 异常。但是，此时线程池不会立刻退出，直到线程池中的任务都已经处理完成，才会退出。

而 shutdownNow()方法是使线程池的状态立刻变成 STOP 状态，并试图停止所有正在执行的线程(如果有if判断则人为地抛出异常)，不再处理还在池队列中等待的任务，当然，它会返回那些未执行的任务。

线程池ThreadPoolExecutor 的拒绝策略：

线程池中的资源全部被占用的时候，对新添加的 Task 任务有不同的处理策略，在默认的情况下，ThreadPoolExecutor 类中有 4 种不同的处理方式:

• AbortPolicy: 当任务添加到线程池中被拒绝时，它将抛出 RejectedExecutionException 异常。
• CallerRunsPolicy : 当任务添加到线程池中被拒绝时，会使用调用线程池的 Thread 线程对象处理被拒绝的任务。
• DiscardOldestPolicy : 当任务添加到线程池中被拒绝时，线程池会放弃等待队列中最旧的未处理任务，然后将被拒绝的任务添加到等待队列中。
• DiscardPolicy: 当任务添加到线程池中被拒绝时，线程池将丢弃被拒绝的任务。

可以通过实现 RejectedExecutionHandler 接口来自定义拒绝策略。RejectedExecutionHandler 接口只包含一个方法 rejectedExecution，该方法在任务无法被执行时被调用。

以下是一个自定义拒绝策略的示例，其中实现了 RejectedExecutionHandler 接口：

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class CustomRejectPolicy implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("Task " + r.toString() + " rejected from " + executor.toString());
        // 在这里实现自定义的拒绝逻辑，比如将被拒绝的任务重新加入队列，或者记录日志等
        // 下面是一个示例 - 捕获到拒绝执行的任务并将其加入到队列中
        try {
            executor.getQueue().put(r);
        } catch (InterruptedException e) {
            // 如果无法再次将任务加入队列，可以进行其他处理，比如记录日志
            System.out.println("Unable to re-queue the task: " + r.toString());
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

然后，可以在创建 ThreadPoolExecutor 实例时，将自定义的拒绝策略传入构造函数：

public class CustomThreadPoolExecutorExample {
    public static void main(String[] args) {
        // 省略其他代码

        RejectedExecutionHandler handler = new CustomRejectPolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, workQueue, threadFactory, handler);

        // 省略其他代码
    }
}
1
2
3
4
5
6
7
8
9
10

在这个示例中，我们创建了一个名为 CustomRejectPolicy 的拒绝策略类，它实现了 RejectedExecutionHandler 接口，并在 rejectedExecution 方法中实现了自定义的拒绝逻辑。然后，在创建 ThreadPoolExecutor 实例时，将该自定义的拒绝策略传入构造函数。

8 Future与Callable

Callable 和 Runnable 都是用于在多线程环境下执行任务的接口，但是它们有几个重要的区别：

1. 返回值：

   • Runnable 的 run 方法没有返回值，因此它通常用于执行没有返回结果的任务。
   • Callable 的 call 方法可以返回执行结果，并且允许抛出受检查异常。我们可以使用 Future 对象来获取 Callable 的执行结果。

2. 异常处理：

   • Runnable 的 run 方法不允许抛出受检查异常，因此如果任务执行过程中发生异常，只能在内部进行处理，不能将异常传递出去。
   • Callable 的 call 方法允许抛出受检查异常，因此允许任务执行过程中抛出异常，调用者可以捕获并处理这些异常。

3. 使用范围：

   • Runnable 是一个函数式接口，可以通过 lambda 表达式来创建实例。
   • Callable 也可以通过 lambda 表达式来创建实例，但是由于它的 call 方法允许抛出受检查异常，需要额外处理异常，因此通常作为返回结果的任务使用。

综合来说，如果你需要在多线程中执行一个没有返回值的任务，你可以使用 Runnable，如果你需要获得任务执行的结果，或者任务可能会抛出异常，你可以使用 Callable。

当使用Callable和Future接口时，可以通过以下示例来执行任务和接收返回值：

import java.util.concurrent.*;

public class CallableFutureExample {
    public static void main(String[] args) {
        // 创建一个线程池
        ExecutorService executor = Executors.newFixedThreadPool(1);

        // 创建一个Callable任务
        Callable<String> task = () -> {
            Thread.sleep(2000); // 模拟一个耗时任务
            return "Hello from Callable!";
        };

        // 提交任务并接收Future对象
        Future<String> future = executor.submit(task);

        // 在这里可以做一些其他的事情，不会阻塞在Future的get方法上

        try {
            // 获取任务的返回值，如果任务还未完成，get方法会阻塞直到任务完成并返回结果
            String result = future.get();
            System.out.println("Task result: " + result);
        } catch (InterruptedException | ExecutionException e) {
            // 处理异常
            e.printStackTrace();
        }

        // 关闭线程池
        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

在这个示例中，我们首先创建了一个 Callable 任务，并且使用线程池提交了这个任务，得到了一个 Future 对象。然后我们可以在 Future 对象上调用 get 方法来等待任务的完成并获得返回结果。在调用 get 方法的过程中，如果任务还未完成，get 方法会阻塞直到任务完成并返回结果。

在主线程中，我们可以在调用 get 方法的同时做一些其他的事情，而不会被阻塞在获取任务结果的操作上。

这种方法允许我们对任务的执行结果进行异步等待和处理，非常灵活和高效。

使用ExecutorService 接口中的方法submit(Runnable,Tresult)方法 submit(Runnable,T result) 的第 2 个参数 result 可以作为执行结果的返回值，而不需要使用 get() 方法来进行获得。
创建实验用的项目futurecallable3，实体类Userinfojava代码如下:

package entity;
public class Userinfo {
privateStringusername;private String password;
public Userinfo() (
super();
public Userinfo(String username，String password)super();
this.username=username;
this.password=password;
//其他set及get方法
}
1
2
3
4
5
6
7
8
9
10

创建类MyRunnable.java代码如下

packagemyrunnable;
import entity.Userinfo;
public class MyRunnable implements Runnable {
private Userinfouserinfo;
public MyRunnable(Userinfo userinfo)super();
this.userinfo = userinfo;
@Override
public void run(){
userinfo.setUsername("usernameValue");userinfo.setPassword("passwordValue");
}}
1
2
3
4
5
6
7
8
9
10

创建类Test.java代码如下:

publicclass Test
FutureTaskabc;
public staticvoidmain(String[] args) (try(Userinfo userinfo = new Userinfo();MyRunnablemyrunnable=new MyRunnable(userinfo);
ThreadPoolExecutor poolnew ThreadPoolExecutor(10，10，10"TimeUnit.SECONDS，new LinkedBlockingDeque<Runnable>());Future<Userinfo> future = pool.submit(myrunnable，userinfo);System.out.printIn("begin time=" +System.currentTimeMillis()userinfo= future.get();System.out.println"get value"+userinfo.getUsername() ++userinfo.getPassword());System.out.println(" end time="+ System.currentTimeMillis());catch (InterruptedException e)[] catch(InterruptedException e)e.printStackTrace();} catch (ExecutionException e) [e.printStackTrace();
1
2
3
4

运行结果：

begin time=1440553015390get value usernameValue passwordValueend time=1440553015390

8.1 方法execute()与submit()的区别

方法 execute() 没有返回值，而 submit()方法可以有返回值。方法 execute()在默认的情况下异常直接抛出，不能捕获，但可以通过自定义 Thread-Factory 的方式进行捕获，而submit()方法在默认的情况下，可以 catch Execution-Exception 捕获异常。

8.2 Future缺点

阻塞：

Callable 接口与 Runnable 接口在对比时主要的优点是，Callable 接口可以通过 Future 取得返回值。但需要注意的是，Future 接口调用 get() 方法取得处理的结果值时是阻塞性的，也就是如果调用Future 对象的 get()方法时，任务尚未执行完成，则调用 get()方法时一直阻塞到此任务完成时为止。如果是这样的效果，则前面先执行的任务一旦耗时很多，则后面的任务调用 get()方法就呈阻塞状态，也就是排队进行等待，大大影响运行效率。也就是主线程并不能保证首先获得的是最先完成任务的返回值，这就是 Future 的缺点，影响效率。

9 CompletionService

在Java中，CompletionService接口没有继承关系，它属于独立的接口。

CompletionService接口用于管理异步任务的执行结果，它允许在任务完成时获取这些任务的结果。它的常见实现类是ExecutorCompletionService，它实现了CompletionService接口，并且可以与Executor框架一起使用。

因此，CompletionService接口没有子类。常见的实现类是ExecutorCompletionService。

CompletionService是Java中的一个接口，位于java.util.concurrent包中。它提供了一种将任务执行结果异步返回的方式，可用于处理并发任务的结果。通常情况下，CompletionService与ExecutorService一起使用，用于提交并执行多个任务，并在任务完成后获取其结果。

以下是一个业务开发中的常见应用代码示例：

import java.util.concurrent.*;

public class CompletionServiceExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        CompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);

        // 提交多个任务
        for (int i = 1; i <= 5; i++) {
            int taskNum = i;
            completionService.submit(new Callable<Integer>() {
                public Integer call() {
                    try {
                        Thread.sleep(1000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return taskNum;
                }
            });
        }

        // 获取任务结果
        for (int i = 0; i < 5; i++) {
            try {
                Future<Integer> result = completionService.take();
                System.out.println("Task " + result.get() + " completed");
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

在这个示例中，我们首先创建了一个ExecutorService，然后使用ExecutorCompletionService来包装它。我们将多个任务提交给CompletionService，并使用take()方法获取任务完成的结果。这种方式可以让我们在任务完成时立即获取其结果，而无需等待所有任务执行完毕。这在需要实时处理并发任务结果的业务场景中非常有用。

使用 CompletionService 接口后，哪个任务先执行完，哪个任务的返回值就先打印。在 CompletionService 接口中如果当前没有任务被执行完，则 completionService.take().get()方法还是呈阻塞特性。

10 ExecutorService

在Java中，ExecutorService接口继承自Executor接口。 ExecutorService扩展了Executor接口，提供了更丰富的任务执行和管理功能。它是用于执行提交的 Callable 对象或者 Runnable 对象的服务。

因此，ExecutorService接口的继承关系如下所示：

Executor
   |
ExecutorService
1
2
3

ExecutorService是一个接口，它定义了一系列用于管理和执行任务的方法，包括执行任务、提交任务、关闭服务等操作。它不能被直接实例化，但可以通过 Executors 类的工厂方法来创建实例。

一些常见的ExecutorService接口的实现类包括：

1. ThreadPoolExecutor：ThreadPoolExecutor是实现了ExecutorService接口的一个线程池类，它可以执行提交的任务，并且提供了一些管理线程池的方法。
2. ScheduledThreadPoolExecutor：ScheduledThreadPoolExecutor是ThreadPoolExecutor的子类，它实现了ScheduledExecutorService接口，可以执行延迟任务和周期性任务。
3. ForkJoinPool：ForkJoinPool是一个用于并行执行任务的线程池，它也实现了ExecutorService接口。
4. AbstractExecutorService：AbstractExecutorService是一个抽象类，它提供了ExecutorService接口的基本实现，可以作为自定义ExecutorService的基类。

ExecutorService是Java中的一个接口，而Executor是一个Java中的接口，用于执行被提交的任务。ExecutorService接口扩展了Executor接口，增加了更丰富的方法和功能，使得可以更方便地管理和控制任务的执行。

Executor接口中只有一个方法execute(Runnable command)，用来执行一个提交的任务。而ExecutorService接口中则包含了一系列提交任务、执行任务、控制任务执行状态等方法，比如submit(Callable task)用来提交有返回结果的任务、shutdown()用来关闭执行器等。

总的来说，我们可以将Executor看作一个执行任务的底层接口，而ExecutorService提供了更多高级功能，可用于管理任务生命周期、获取任务执行结果、控制执行器状态等。ExecutorService是在Executor接口的基础上进行了扩展，使得任务执行更加方便、灵活。

10.1 invokeAny()与invokeAll()

ExecutorService接口中的invokeAny()和invokeAll()方法用于执行一组任务，并返回它们的执行结果。这两个方法可以用于并发地执行多个任务，并在所有任务完成后获取它们的结果。

1. invokeAny()方法：
   • 当一组任务中的任何一个任务成功完成（没有抛出异常），invokeAny()方法将立即返回这个任务的执行结果，并取消所有其他任务的执行。如果没有任务成功完成，该方法将抛出ExecutionException异常。
   • 方法签名如下：

   <T> T invokeAny(Collection<? extends Callable<T>> tasks)
           throws InterruptedException, ExecutionException;
   1
   2

以下是一个使用invokeAny()方法的示例代码：

ExecutorService executor = Executors.newFixedThreadPool(5);
Set<Callable<String>> tasks = new HashSet<>();
tasks.add(() -> "Task 1");
tasks.add(() -> "Task 2");
tasks.add(() -> "Task 3");

String result = executor.invokeAny(tasks);
System.out.println("Result: " + result);
executor.shutdown();
1
2
3
4
5
6
7
8
9

2. invokeAll()方法：
   • invokeAll()方法会执行一组任务，并等待所有任务完成后返回它们的执行结果。如果所有任务成功完成，方法将返回一个包含所有任务执行结果的列表。如果任何一个任务抛出异常，该方法将立即返回，并抛出ExecutionException异常。
   • 方法签名如下：

   <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
   1
   2

以下是一个使用invokeAll()方法的示例代码：

ExecutorService executor = Executors.newFixedThreadPool(5);
Set<Callable<String>> tasks = new HashSet<>();
tasks.add(() -> "Task 1");
tasks.add(() -> "Task 2");
tasks.add(() -> "Task 3");

List<Future<String>> results = executor.invokeAll(tasks);
for (Future<String> result : results) {
    System.out.println("Result: " + result.get());
}
executor.shutdown();
1
2
3
4
5
6
7
8
9
10
11

在以上示例中，我们使用ExecutorService的invokeAny()和invokeAll()方法执行一组任务，并获取它们的执行结果。这两个方法非常适合用于并发任务的执行和获取结果。

11 ScheduledExecutorService

类ScheduledExecutorService的主要作用就是可以将定时任务与线程池功能结合使用。

在Java中，ScheduledExecutorService接口继承自ExecutorService接口。ScheduledExecutorService是用于延迟执行任务或者周期性执行任务的接口，它在ExecutorService的基础上增加了一些用于延迟执行和周期性执行任务的方法。

因此，ScheduledExecutorService接口的继承关系如下所示：

Executor
   |
ExecutorService
   |
ScheduledExecutorService
1
2
3
4
5

ScheduledExecutorService接口的一些常见的实现类包括：

1. ThreadPoolExecutor：ThreadPoolExecutor是实现了ExecutorService接口的一个线程池类，它可以执行提交的任务。ScheduledThreadPoolExecutor是ThreadPoolExecutor的子类，实现了ScheduledExecutorService接口，可以执行延迟任务和周期性任务。

2. ForkJoinPool：ForkJoinPool是一个用于并行执行任务的线程池，它实现了ExecutorService接口。ForkJoinPool也实现了ScheduledExecutorService接口，因此它可以处理延迟任务和周期性任务。

ScheduledExecutorService 是一个接口，它是 ExecutorService 的子接口，用于执行延迟任务或者周期性任务。它提供了一些方法来安排任务在给定延迟之后执行，或者以固定速率执行。ScheduledExecutorService 提供了比 Timer 更灵活的调度任务的方法，并且可以同时执行多个任务。

下面是一个简单的示例，演示了如何使用 ScheduledExecutorService 来执行延迟任务和周期性任务：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledExecutorServiceExample {
    public static void main(String[] args) {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

        // 延迟任务，5秒后执行
        executor.schedule(() -> System.out.println("Delayed task executed"), 5, TimeUnit.SECONDS);

        // 周期性任务，每隔3秒执行一次
        executor.scheduleAtFixedRate(() -> System.out.println("Periodic task executed"), 0, 3, TimeUnit.SECONDS);

        // 等待一定时间以便让任务执行
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 关闭执行器
        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

在这个示例中，我们使用 Executors.newScheduledThreadPool() 方法创建了一个 ScheduledExecutorService。然后，我们使用 schedule() 方法安排了一个延迟任务，在5秒后执行。接着，使用 scheduleAtFixedRate() 方法安排了一个每隔3秒执行一次的周期性任务。最后，我们让主线程休眠10秒以等待任务执行，然后调用 shutdown() 方法关闭执行器。

总的来说，ScheduledExecutorService 提供了一种方便的方式来执行延迟任务和周期性任务，并且相比于 Timer 更加灵活和可靠。

任务既可以是Callable也可以是Runnable。

11.1 Executors.newSingleThreadscheduledExecutor()

Executors.newSingleThreadScheduledExecutor() 是一个工厂方法，用于创建一个具有单个线程的 ScheduledExecutorService。ScheduledExecutorService 是一个能够在给定的延迟之后或者周期性地执行任务的 ExecutorService。

下面是一个代码示例，演示了如何使用 newSingleThreadScheduledExecutor() 方法：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class SingleThreadScheduledExecutorExample {
    public static void main(String[] args) {
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

        Runnable task = () -> {
            System.out.println("Executing task at " + System.currentTimeMillis());
        };

        // 在指定延迟后执行任务
        executor.schedule(task, 5, TimeUnit.SECONDS);

        // 每隔一定的时间执行任务
        executor.scheduleAtFixedRate(task, 0, 3, TimeUnit.SECONDS);

        // 关闭执行器
        executor.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

在这个示例中，我们首先调用 Executors.newSingleThreadScheduledExecutor() 方法创建了一个单线程的 ScheduledExecutorService。然后，我们创建了一个任务（Runnable），并且使用 schedule() 方法在5秒后执行该任务，并使用 scheduleAtFixedRate() 方法让该任务每隔3秒执行一次。最后，我们调用 shutdown() 方法来关闭执行器。

这个例子展示了如何使用newSingleThreadScheduledExecutor()方法创建一个具有单个线程的 ScheduledExecutorService，并安排任务在未来的某个时间点执行，或者以固定的速率周期性地执行任务。

11.2 scheduleAtFixedRate()

scheduleAtFixedRate() 方法是 ScheduledExecutorService 接口中的一个方法，用于安排周期性任务以一定的速率执行。该方法可以按照固定的时间间隔执行任务，即使前一个任务还没有执行完。

方法签名如下：

ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
1

• command 参数是要执行的任务，通常是一个实现了 Runnable 接口的任务。
• initialDelay 参数是任务开始执行之前的延迟时间。
• period 参数是两次任务之间的时间间隔。
• unit 参数是时间单位。

下面是一个示例，演示了如何使用 scheduleAtFixedRate() 方法执行一个周期性任务：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduleAtFixedRateExample {
    public static void main(String[] args) {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

        Runnable task = () -> {
            System.out.println("Executing task at " + System.currentTimeMillis());
        };

        // 在初始延迟0秒后，每隔3秒执行一次任务
        executor.scheduleAtFixedRate(task, 0, 3, TimeUnit.SECONDS);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

在这个例子中，我们使用 scheduleAtFixedRate() 方法安排了一个周期性任务，该任务会在初始延迟0秒后，每隔3秒执行一次。当程序运行时，该任务将会以指定的速率执行，即使任务执行时间超过了间隔时间，新的任务也会按照规定执行。

11.3 scheduleWithFixedDelay()

scheduleWithFixedDelay() 方法是 ScheduledExecutorService 接口中的一个方法，用于安排周期性任务以一定的延迟执行。不同于 scheduleAtFixedRate() 方法，scheduleWithFixedDelay() 方法会等待前一个任务完成后的指定延迟时间后再执行下一个任务。

方法签名如下：

ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
1

• command 参数是要执行的任务，通常是一个实现了 Runnable 接口的任务。
• initialDelay 参数是任务开始执行之前的延迟时间。
• delay 参数是前一个任务结束和下一个任务开始之间的时间间隔。
• unit 参数是时间单位。

下面是一个示例，演示了如何使用 scheduleWithFixedDelay() 方法执行一个周期性任务 with fixed delay：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduleWithFixedDelayExample {
    public static void main(String[] args) {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

        Runnable task = () -> {
            System.out.println("Executing task at " + System.currentTimeMillis());
            try {
                Thread.sleep(1000); // 模拟任务执行时间
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        // 在初始延迟0秒后，每次任务执行完成后，延迟2秒再执行
        executor.scheduleWithFixedDelay(task, 0, 2, TimeUnit.SECONDS);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

在这个例子中，我们使用 scheduleWithFixedDelay() 方法安排了一个周期性任务，该任务会在初始延迟0秒后开始执行，每次任务执行完成后，会等待2秒后再执行下一个任务。当程序运行时，我们可以看到任务按照指定的延迟时间周期性执行。

12 Fork Join分治

在JDK中并行执行框架Fork-Join使用了“工作窃取(work-stealing”算法，它是指某个线程从其他队列里窃取任务来执行，那这样做有什么优势或者目的是什么呢?比如要完成一个比较大的任务，完全可以把这个大的任务分割为若干互不依赖的子任务/小任务，为了更加方便地管理这些任务，于是把这些子任务分别放到不同的队列里，这时就会出现有的线程会先把自己队列里的任务快速执行完毕，而其他线程对应的队列里还有任务等待处理，完成任务的线程与其等着，不如去帮助其他线程分担要执行的任务，于是它就去其他线程的队列里窃取一个任务来执行，这就是所谓的“工作窃取(work-stealing)”算法。

Fork-Join 框架是 Java 并发库中的一部分，用于实现并行计算。它使用递归任务分割（work-stealing）的方法来提高多核处理器上的任务执行效率。

Fork-Join 框架的核心类是 ForkJoinPool 和 RecursiveTask 或 RecursiveAction。

• ForkJoinPool 是一个线程池的扩展，它管理工作线程以执行 Fork-Join 任务。
• RecursiveTask 和 RecursiveAction 分别是用于有返回值和没有返回值的任务。它们是可以递归拆分的任务类型。

下面是一个简单的示例，演示了如何使用 Fork-Join 框架来并行计算数组中元素的和：

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinExample {
    private static class SumTask extends RecursiveTask<Integer> {
        private static final int THRESHOLD = 5; // 任务拆分的阈值
        private int[] array;
        private int start;
        private int end;

        public SumTask(int[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }

        protected Integer compute() {
            if (end - start <= THRESHOLD) {
                int sum = 0;
                for (int i = start; i < end; i++) {
                    sum += array[i];
                }
                return sum;
            } else {
                int mid = (start + end) >>> 1;
                SumTask leftTask = new SumTask(array, start, mid);
                SumTask rightTask = new SumTask(array, mid, end);
                leftTask.fork();
                int rightResult = rightTask.compute();
                int leftResult = leftTask.join();
                return leftResult + rightResult;
            }
        }
    }

    public static void main(String[] args) {
        int[] array = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        SumTask task = new SumTask(array, 0, array.length);
        ForkJoinPool pool = new ForkJoinPool();
        int result = pool.invoke(task);
        System.out.println("Sum: " + result);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

在这个示例中，我们创建了一个 SumTask 类，用于计算数组中元素的和。当任务范围大于阈值时，任务会被拆分成两个子任务并由 fork() 方法提交到 Fork-Join 框架中执行，然后使用 join() 方法等待子任务的结果，最终将结果合并返回。最后，我们通过 ForkJoinPool 类来执行任务并获取任务的结果。

当代码运行时，Fork-Join 框架将会自动通过多线程并行执行任务，从而提高任务计算的效率。

这段代码是一个典型的分治算法的实现，用于将数组中的元素拆分成两个子任务进行并行计算。让我逐步解释其中的意思：

1. int mid = (start + end) >>> 1; - 这一行计算出数组范围的中间位置，用于将数组拆分成两个子数组。

2. SumTask leftTask = new SumTask(array, start, mid); - 这一行创建了一个新的 SumTask 任务，用于计算左半部分数组的和。

3. SumTask rightTask = new SumTask(array, mid, end); - 这一行创建了另一个 SumTask 任务，用于计算右半部分数组的和。

4. leftTask.fork(); - 这一行使用 fork 方法提交左半部分数组的计算任务到 Fork-Join 框架中进行执行。

5. int rightResult = rightTask.compute(); - 这一行直接在当前线程中计算右半部分数组的和，因为左半部分数组的任务已经使用 fork 方法提交到了 Fork-Join 框架中。

6. int leftResult = leftTask.join(); - 这一行使用 join 方法获取左半部分数组计算任务的结果。如果计算任务还没有完成，则会等待其完成。

7. return leftResult + rightResult; - 最后，将左半部分数组的计算结果和右半部分数组的计算结果相加，然后作为当前任务的计算结果返回。

总的来说，这段代码实现了一个并行计算数组和的任务，通过拆分数组并使用 Fork-Join 框架实现多线程并行计算，从而提高计算效率。

在 ForkJoinPool.java 类中的 execute() 方法是以异步的方式执行任务

13 并发集合框架

13.1 Vector

在Java中，Vector是一个基于数组实现的动态数组，它与ArrayList非常相似，但Vector是线程安全的，因为它的所有方法都是同步的。但由于这种同步，Vector的性能可能不如ArrayList。在大多数情况下，推荐使用ArrayList，除非需要线程安全性。

下面是一个简单的Vector的使用示例：

import java.util.Vector;

public class VectorExample {
    public static void main(String[] args) {
        Vector<String> vector = new Vector<>();

        // 添加元素
        vector.add("Apple");
        vector.add("Banana");
        vector.add("Orange");

        // 获取元素
        System.out.println("Element at index 1: " + vector.get(1));

        // 修改元素
        vector.set(2, "Mango");

        // 删除元素
        vector.remove(0);

        // 遍历元素
        for (String fruit : vector) {
            System.out.println(fruit);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

在这个示例中，我们创建了一个Vector对象并向其中添加了一些元素。然后我们使用get()方法获取特定索引处的元素，使用set()方法修改元素，使用remove()方法删除元素，并用for-each循环遍历输出每个元素。

需要注意的是，Vector作为一个线程安全的集合，添加、删除、获取元素的操作会比ArrayList略慢，因此在大部分情况下推荐使用ArrayList。如果线程安全性不是关键问题，而同时需要可变大小的数组，推荐优先使用ArrayList。

13.2 Stack

在Java中，Stack类表示了一个后进先出（LIFO）的对象集合。它继承自Vector类，但提供了对堆栈顶部的元素进行压入（push）和弹出（pop）的基本操作。Stack类在Java中已经被标记为遗留类（Legacy Class），通常在实际开发中更推荐使用Deque接口的实现类（比如ArrayDeque），因为Deque既提供了栈的功能，也提供了队列的功能，并且性能更好。

下面是一个简单的Stack类的使用示例：

import java.util.Stack;

public class StackExample {
    public static void main(String[] args) {
        Stack<String> stack = new Stack<>();

        // 压入元素
        stack.push("A");
        stack.push("B");
        stack.push("C");

        // 弹出元素
        String poppedElement = stack.pop();
        System.out.println("Popped element: " + poppedElement);

        // 获取堆栈顶部元素
        String topElement = stack.peek();
        System.out.println("Top element: " + topElement);

        // 判断堆栈是否为空
        System.out.println("Is stack empty? " + stack.isEmpty());

        // 遍历元素
        System.out.println("Elements in stack: ");
        for (String element : stack) {
            System.out.println(element);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

在这个示例中，我们创建了一个Stack对象并向其中压入了一些元素。然后我们使用pop()方法弹出栈顶元素，使用peek()方法获取栈顶元素但不移除它，并使用isEmpty()方法判断堆栈是否为空。最后，我们使用for-each循环遍历输出每个元素。

需要注意的是，由于Stack类是Vector的子类，因此它也继承了Vector的同步特性，但由于这种同步会导致性能下降，因此在实际开发中建议使用Deque接口的实现类来代替Stack。

13.3 Deque

接口Queue可以支持对表头的操作，而接口 Deque不仅支持对表头进行操作，而且还支持对表尾进行操作，所以Dequq的全称为“double ended queue(双端队列)”。

接口Deque的非并发实现类有ArrayDeque和LinkedList，它们之间有一些区别，如果只想实现从队列两端获取数据则使用ArrayDeque如果想实现从队列两端获取数据时还可以根据索引的位置操作数据则使用 LinkedList。

在Java中，可以使用ConcurrentLinkedDeque来实现线程安全的Deque。ConcurrentLinkedDeque是Deque接口的一个实现，它提供了线程安全的操作，适用于多线程并发访问的场景。

ConcurrentLinkedDeque的特点包括：

• 它是非阻塞的，并发性能良好，能够支持并发的插入、删除和遍历操作。
• 它不保证元素的排序。
• 此外，ConcurrentLinkedDeque还提供了一些其他并发安全的方法，比如offer、poll、peek等操作。

下面是一个简单的使用示例：

import java.util.Deque;
import java.util.concurrent.ConcurrentLinkedDeque;

public class ConcurrentLinkedDequeExample {
    public static void main(String[] args) {
        Deque<String> deque = new ConcurrentLinkedDeque<>();

        deque.offer("A");
        deque.offer("B");

        System.out.println("Deque elements: " + deque);

        String element = deque.poll();
        System.out.println("Popped element: " + element);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

在这个示例中，我们创建了一个ConcurrentLinkedDeque对象，并使用offer方法向队列中添加元素，然后使用poll方法从队列中弹出一个元素。ConcurrentLinkedDeque能够确保这些操作在多线程环境下的安全性。

13.4 Set

接口Set也是对Collection 接口进行了扩展，它具有的默认特点是内容不允许重复，排序方式为自然排序，防止元素重复的原理是元素需要重写 hashCode0和equals0方法。

接口Set 最常用的不支持并发的实现类就是 HashSet。HashSet 默认以无序的方式组织元素，而 LinkedHashSet类可以有序的组织元素。

接口Set还有另外一个实现类，名称为 TreeSet，它不仅实现了 Set接口，而且还实现了SortedSet 和NavigableSet 接口，而 SortedSet 接口的父接口为 Set，

SortedSet和 NavigableSet接口在功能上得到了扩展，比如可以获取 Set 中内容的子集，以比较范围进行获得子集，支持对表头与表尾的数据进行获取等。

LinkedHashSet和TreeSet都不是线程安全的。它们是基于哈希表和树结构的集合实现，它们的操作并不是同步的。这意味着如果多个线程同时访问或修改LinkedHashSet或TreeSet的内容，可能会引发并发问题，比如数据不一致或意外的结果。

如果需要线程安全的集合，你可以考虑使用Collections类中的工具方法对现有的非线程安全的集合进行包装，或者使用并发集合类，例如ConcurrentLinkedHashSet或ConcurrentSkipListSet。这些并发集合类被设计用来在并发环境下安全地进行读写操作。

13.5 非阻塞队列

非阻塞队列的特色就是队列里面没有数据时，操作队列出现异常或返回 nul1，不具有等待/阻塞的特色。
在JDK的并发包中，常见的非阻塞队列有:

1. ConcurrentHashMap
2. ConcurrentSkipListMap（ConcurrentHashMap 不支持排序，虽然 LinkedHashMap 支持 key的顺序性，但又不支持并发，那么如果出现这种既要求并发安全性，而又要求排序的情况就可以使用类 ConcurrentSkipListMap。）
3. ConcurrentSkipListSet（支持排序而且不允许重复的元素）
4. ConcurrentLinkedQueue（头操作）
5. ConcurrentLinkedDeque（头尾操作）
6. CopyOnWriteArrayList（线程安全List）
7. CopyOnWriteArraySet（线程安全Set）

13.6 HashTable

HashTable是线程安全的。在多线程环境下，HashTable的所有公共方法都是同步的，因此可以在多个线程中安全地进行读取和写入操作。这意味着多个线程可以同时访问和修改HashTable的内容而不会导致数据不一致或其他并发问题。

然而，尽管HashTable是线程安全的，它并不推荐在现代Java应用中使用。这是因为它的同步机制可能会导致性能下降，在大多数情况下，使用ConcurrentHashMap会更加高效。ConcurrentHashMap提供了与HashTable类似的线程安全特性，但具有更好的性能。

另外需要注意的是，HashTable不允许空键或空值，当试图插入空键或空值时，将会抛出NullPointerException异常。因此，通常更推荐使用ConcurrentHashMap或其他基于Map接口的线程安全实现。

HashTable和ConcurrentHashMap都是用于实现键值对存储的数据结构，它们都可以在多线程环境中安全地进行并发访问。但是，它们在实现和性能上有一些重要的区别。

1. 线程安全性：

   • HashTable：HashTable的所有公共方法都是同步的，因此可以在多个线程中安全地进行读取和写入操作。
   • ConcurrentHashMap：ConcurrentHashMap使用了一种更加精细的锁机制，它允许多个读操作同时进行，而不会阻塞，因此在读多写少的情况下具有更好的性能。

2. 性能：

   • HashTable：由于所有方法都是同步的，对于大多数操作，HashTable的性能通常会比较低，在高并发情况下可能会出现性能瓶颈。
   • ConcurrentHashMap：ConcurrentHashMap通过精心设计的并发控制机制，能够提供更好的并发性能，尤其在读操作较多的情况下能够比较好地提高性能。

3. 空键值：

   • HashTable：不允许空键或空值，插入空键或空值时会抛出NullPointerException异常。
   • ConcurrentHashMap：允许空键和空值的存在。

4. 迭代器：

   • HashTable：其迭代器是Fail-Fast的，即在迭代中若有其他线程修改了HashTable结构，会立即抛出ConcurrentModificationException异常。
   • ConcurrentHashMap：其迭代器是Weakly Consistent的，允许在遍历的过程中进行修改，并不会立即抛出异常。

综上所述，虽然HashTable是线程安全的，但在现代Java应用中，通常更推荐使用ConcurrentHashMap来实现线程安全的并发存储，因为ConcurrentHashMap在性能和灵活性上都优于HashTable。

其实主要的差异就是 Hashtable 不支持在循环中 remove元素。

13.7 阻塞队列

在JDK 中提供了若干集合工具类都具有阻塞特性，所谓的阻塞队列 BlockingQueue，其实就是如果 BlockQueue 是空的，从 BlockingQueue 取东西的操作将会被阻塞进人等待状态直到 BlockingQueue 添加进了元素才会被唤醒。同样，如果 BlockingOueue 是满的，也就是没有空余空间时，试图往队列中存放元素的操作也会被阻塞进入等待状态，直到 BlockingQueue里有剩余空间才会被唤醒继续操作。

• ArrayBlockingQueue（支持并发有界阻塞队列）
• PriorityBlockingQueue（支持并发的优先级队列）
• LinkedBlockingQueue（支持并发无界阻塞队列）
• SynchronousQueue

SynchronousQueue是Java中的一个特殊类型的阻塞队列，它的特点是只能容纳单个元素。与其他阻塞队列不同，SynchronousQueue在没有消费者线程的情况下，插入元素会被阻塞；在没有生产者线程的情况下，取出元素也会被阻塞。它通常用于线程间的直接传输，其中一个线程将某个对象交给另一个线程。

下面是一个简单的Java代码示例，演示了SynchronousQueue的基本用法：

import java.util.concurrent.*;

public class SynchronousQueueExample {
    public static void main(String[] args) {

        SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();

        // 生产者线程
        new Thread(() -> {
            try {
                String element = "Hello, SynchronousQueue!";
                synchronousQueue.put(element); // 将元素放入队列
                System.out.println("Produced: " + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 消费者线程
        new Thread(() -> {
            try {
                String element = synchronousQueue.take(); // 从队列中取出元素
                System.out.println("Consumed: " + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

在这个示例中，我们创建了一个SynchronousQueue，并启动了生产者和消费者线程。在生产者线程中，我们使用put()方法向队列中放入一个元素；在消费者线程中，我们使用take()方法从队列中取出元素。由于SynchronousQueue的特性，生产者线程会在put()方法处被阻塞，直到消费者线程调用take()方法取出元素为止。

总之，SynchronousQueue是一个特殊的阻塞队列，用于在线程之间进行直接传输元素。它的特点是只能容纳单个元素，并且在插入和取出元素时会阻塞线程。

• DelayQueue

DelayQueue是Java中的一个阻塞队列，它用于存放实现了Delayed接口的元素。队列中的元素只有在其延迟期满后才能被取出。DelayQueue通常用于实现定时任务调度以及超时处理。

下面是一个简单的Java代码示例，演示了DelayQueue的基本用法：

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

class DelayedElement implements Delayed {
    private String data;
    private long expireTime;

    public DelayedElement(String data, long delay) {
        this.data = data;
        this.expireTime = System.currentTimeMillis() + delay;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = expireTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        long diff = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        return (int) diff;
    }

    public String getData() {
        return data;
    }
}

public class DelayQueueExample {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedElement> delayQueue = new DelayQueue<>();

        // 添加元素到队列
        delayQueue.put(new DelayedElement("Task 1", 1000));
        delayQueue.put(new DelayedElement("Task 2", 5000));
        delayQueue.put(new DelayedElement("Task 3", 2000));

        // 从队列中取出元素
        while (!delayQueue.isEmpty()) {
            DelayedElement element = delayQueue.take();
            System.out.println("Processing: " + element.getData());
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

在这个示例中，我们创建了一个DelayQueue，并向其中添加了三个实现了Delayed接口的延迟元素。每个元素代表一个任务，并且有一个延迟时间。我们使用put()方法将元素放入队列，在取出元素时，只有在其延迟期满后才能被取出。

DelayQueue对元素的处理是受到延迟时间控制的，例如，如果某个元素的延迟时间还未到期，那么take()方法将会一直阻塞。

总之，DelayQueue是一个用于存放实现了Delayed接口的元素的阻塞队列，它通常用于实现定时任务调度以及超时处理。

• LinkedTransferQueue

LinkedTransferQueue是Java中的一个特殊类型的并发队列，它既可以作为阻塞队列使用，也可以作为同步队列使用。它是一个无界的队列，可用于实现生产者-消费者模式以及异步任务处理。

下面是一个简单的Java代码示例，演示了LinkedTransferQueue的基本用法：

import java.util.concurrent.LinkedTransferQueue;

public class LinkedTransferQueueExample {
    public static void main(String[] args) throws InterruptedException {
        LinkedTransferQueue<String> transferQueue = new LinkedTransferQueue<>();

        // 生产者线程
        new Thread(() -> {
            try {
                String element = "Hello, LinkedTransferQueue!";
                transferQueue.transfer(element); // 等待元素被消费
                System.out.println("Produced: " + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 消费者线程
        new Thread(() -> {
            try {
                String element = transferQueue.take(); // 获取元素
                System.out.println("Consumed: " + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

在这个示例中，我们创建了一个LinkedTransferQueue，并启动了生产者和消费者线程。在生产者线程中，我们使用transfer()方法向队列中传输一个元素，并等待该元素被消费；在消费者线程中，我们使用take()方法从队列中获取元素。

LinkedTransferQueue的一个特性是，生产者线程在使用transfer()方法时会一直等待，直到有消费者线程来获取元素；而消费者线程在使用take()方法时，如果队列中有元素，将立即取出并继续执行，否则会阻塞等待元素到来。

总之，LinkedTransferQueue是一个特殊的并发队列，既可以作为阻塞队列使用，也可以作为同步队列使用。它适用于实现生产者-消费者模式和异步任务处理。