Java 编程问题:十、并发-线程池、可调用对象和同步器2

简介: Java 编程问题:十、并发-线程池、可调用对象和同步器

Java 编程问题:十、并发-线程池、可调用对象和同步器1https://developer.aliyun.com/article/1426161

201 Java 中的线程池

线程池是可用于执行任务的线程的集合。线程池负责管理其线程的创建、分配和生命周期,并有助于提高性能。现在,我们来谈谈遗嘱执行人。

Executor

java.util.concurrent包中,有一堆专用于执行任务的接口。最简单的一个叫做Executor。这个接口公开了一个名为execute(Runnable command)的方法。下面是使用此方法执行单个任务的示例:

public class SimpleExecutor implements Executor {
  @Override
  public void execute(Runnable r) {
    (new Thread(r)).start();
  }
}
SimpleExecutor se = new SimpleExecutor();
se.execute(() -> {
  System.out.println("Simple task executed via Executor interface");
});

ExecutorService

一个更复杂、更全面的接口提供了许多附加方法,它是ExecutorService。这是Executor的丰富版本。Java 附带了一个成熟的实现ExecutorService,名为ThreadPoolExecutor。这是一个线程池,可以用一组参数实例化,如下所示:

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

下面是对前面代码中实例化的每个参数的简短描述:

  • corePoolSize:池中要保留的线程数,即使它们是空闲的(除非设置了allowCoreThreadTimeOut
  • maximumPoolSize:允许的最大线程数
  • keepAliveTime:当这个时间过去后,空闲线程将从池中移除(这些是超过corePoolSize的空闲线程)
  • unitkeepAliveTime参数的时间单位
  • workQueue:在Runnable的实例(只有execute()方法提交的Runnable任务)执行之前,用来存放这些实例的队列
  • threadFactory:执行器创建新线程时使用此工厂
  • handler:当ThreadPoolExecutor由于饱和而无法执行Runnable时,即线程边界和队列容量已满(例如,workQueue大小固定,同时设置了maximumPoolSize),它将控制和决策交给这个处理器

为了优化池大小,我们需要收集以下信息:

  • CPU 数量(Runtime.getRuntime().availableProcessors()
  • 目标 CPU 利用率(在范围内,[0, 1]
  • 等待时间(W)
  • 计算时间(C)

以下公式有助于我们确定池的最佳大小:

Number of threads 
  = Number of CPUs * Target CPU utilization * (1 + W/C)

根据经验,对于计算密集型任务(通常是小型任务),最好使用线程数等于处理器数或处理器数 +1(以防止潜在的暂停)来对线程池进行基准测试。对于耗时且阻塞的任务(例如,I/O),更大的池更好,因为线程将无法以高速率进行调度。另外,还要注意与其他池(例如,数据库连接池和套接字连接池)的干扰。

让我们看一个ThreadPoolExecutor的例子:

public class SimpleThreadPoolExecutor implements Runnable {
  private final int taskId;
  public SimpleThreadPoolExecutor(int taskId) {
    this.taskId = taskId;
  }
  @Override
  public void run() {
    Thread.sleep(2000);
    System.out.println("Executing task " + taskId 
      + " via " + Thread.currentThread().getName());
  }
  public static void main(String[] args) {
    BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(5);
    final AtomicInteger counter = new AtomicInteger();
    ThreadFactory threadFactory = (Runnable r) -> {
      System.out.println("Creating a new Cool-Thread-" 
        + counter.incrementAndGet());
      return new Thread(r, "Cool-Thread-" + counter.get());
    };
    RejectedExecutionHandler rejectedHandler
      = (Runnable r, ThreadPoolExecutor executor) -> {
        if (r instanceof SimpleThreadPoolExecutor) {
          SimpleThreadPoolExecutor task=(SimpleThreadPoolExecutor) r;
          System.out.println("Rejecting task " + task.taskId);
        }
    };
    ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 20, 1,
      TimeUnit.SECONDS, queue, threadFactory, rejectedHandler);
    for (int i = 0; i < 50; i++) {
      executor.execute(new SimpleThreadPoolExecutor(i));
    }
    executor.shutdown();
    executor.awaitTermination(
      Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
  }
}

main()方法触发Runnable50 个实例。每个Runnable睡两秒钟,打印一条消息。工作队列限制为Runnable的 5 个实例——核心线程为 10,线程数最多为 20 个,空闲超时为 1 秒。可能的输出如下:

Creating a new Cool-Thread-1
...
Creating a new Cool-Thread-20
Rejecting task 25
...
Rejecting task 49
Executing task 22 via Cool-Thread-18
...
Executing task 12 via Cool-Thread-2

ScheduledExecutorService

ScheduledExecutorService是一个ExecutorService,它可以安排任务在给定的延迟后执行,或者定期执行。这里,我们有schedule()scheduleAtFixedRate()scheduleWithFixedDelay()等方法。schedule()用于一次性任务,scheduleAtFixedRate()scheduleWithFixedDelay()用于周期性任务。

通过执行器的线程池

更进一步,我们将介绍辅助类Executors。此类使用以下方法公开几种类型的线程池:

  • newSingleThreadExecutor():这是一个线程池,只管理一个线程,队列无限,一次只执行一个任务:
ExecutorService executor 
  = Executors.newSingleThreadExecutor();
  • newCachedThreadPool():这是一个线程池,根据需要创建新线程并删除空闲线程(60 秒后);核心池大小为 0,最大池大小为Integer.MAX_VALUE(此线程池在需求增加时扩展,在需求减少时收缩):
ExecutorService executor = Executors.newCachedThreadPool();
  • newFixedThreadPool():这是一个线程数固定、队列无限的线程池,产生无限超时的效果(核心池大小和最大池大小等于指定的大小):
ExecutorService executor = Executors.newFixedThreadPool(5);
  • newWorkStealingThreadPool():这是一个基于工作窃取算法的线程池(它充当 Fork/Join 框架上的一层):
ExecutorService executor = Executors.newWorkStealingPool();
  • newScheduledThreadPool():一个线程池,可以安排命令在给定的延迟后运行,或者定期执行(我们可以指定核心池的大小):
ScheduledExecutorService executor 
  = Executors.newScheduledThreadPool(5);

202 具有单个线程的线程池

为了演示单线程线程池的工作原理,假设我们想编写一个程序,模拟装配线(或输送机),用两个工作器检查和包装灯泡。

通过检查,我们了解到工作器测试灯泡是否亮起。通过包装,我们了解到,工作器将经过验证的灯泡拿到盒子里。这种工艺在几乎所有工厂都很常见。

两名工作器如下:

  • 一种所谓的生产者(或检查者),负责测试每个灯泡,看灯泡是否亮起
  • 一个所谓的消费者(或包装商),负责将每个检查过的灯泡包装到一个盒子里

这种问题完全适合于下图所示的生产者消费者设计模式:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wMW6fHCN-1657345732716)(https://github.com/apachecn/apachecn-java-zh/raw/master/docs/java-coding-prob/img/817c7786-9093-401c-9afd-bf9b39aabd25.png)]

最常见的是,在这种模式中,生产者和消费者通过队列(生产者排队数据,消费者排队数据)进行通信。这个队列称为数据缓冲区。当然,根据流程设计,其他数据结构也可以起到数据缓冲的作用。

现在,让我们看看如果生产者等待消费者可用,我们如何实现这个模式。

稍后,我们将为不等待消费者的生产者实现此模式。

生产者等待消费者出现

装配线启动时,生产者将逐个检查进线灯泡,而消费者将将其打包(每个盒子中有一个灯泡)。此流重复,直到装配线停止。

下图是生产者消费者之间的流程图:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-yPBj7z7O-1657345732716)(https://github.com/apachecn/apachecn-java-zh/raw/master/docs/java-coding-prob/img/3cf100e5-7ad8-4548-9a33-1e55a146b655.png)]

我们可以将装配线视为我们工厂的助手,因此它可以实现为助手或工具类(当然,它也可以很容易地切换到非static实现,因此如果对您的情况更有意义,请随意切换):

public final class AssemblyLine {
  private AssemblyLine() {
    throw new AssertionError("There is a single assembly line!");
  }
  ...
}

当然,实现这个场景的方法很多,但是我们对使用 JavaExecutorService感兴趣,更准确地说是Executors.newSingleThreadExecutor()。使用一个工作线程来操作未绑定队列的Executor通过此方法创建。

我们只有两个工作器,所以我们可以使用两个实例Executor(一个Executor将启动生产者,另一个将启动消费者)。因此,生产者将是一个线程,消费者将是另一个线程:

private static ExecutorService producerService;
private static ExecutorService consumerService;

由于生产者和消费者是好朋友,他们决定根据一个简单的场景来工作:

  • 只有消费者不忙时,生产者才会检查灯泡并将其传递给消费者(如果消费者忙,生产者会等待一段时间,直到消费者有空)
  • 生产者在将当前灯泡传递给用户之前不会检查下一个灯泡
  • 消费者将尽快包装每个进入的灯泡

这个场景适用于TransferQueueSynchronousQueue,它执行的过程与前面提到的场景非常相似。让我们使用TransferQueue。这是一个BlockingQueue,其中生产者可以等待消费者接收元素。BlockingQueue实现是线程安全的:

private static final TransferQueue<String> queue 
  = new LinkedTransferQueue<>();

生产者和消费者之间的工作流程是先进先出FIFO)类型:第一个检查的灯泡是第一个打包的灯泡),因此LinkedTransferQueue是一个不错的选择。

一旦装配线启动,生产者将持续检查灯泡,因此我们可以将其作为一个类来实现,如下所示:

private static final int MAX_PROD_TIME_MS = 5 * 1000;
private static final int MAX_CONS_TIME_MS = 7 * 1000;
private static final int TIMEOUT_MS = MAX_CONS_TIME_MS + 1000;
private static final Random rnd = new Random();
private static volatile boolean runningProducer;
...
private static class Producer implements Runnable {
  @Override
  public void run() {
    while (runningProducer) {
      try {
        String bulb = "bulb-" + rnd.nextInt(1000);
        Thread.sleep(rnd.nextInt(MAX_PROD_TIME_MS));
        boolean transfered = queue.tryTransfer(bulb,
          TIMEOUT_MS, TimeUnit.MILLISECONDS);
        if (transfered) {
          logger.info(() -> "Checked: " + bulb);
        }
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        logger.severe(() -> "Exception: " + ex);
        break;
      }
    }
  }
}

因此,生产者通过tryTransfer()方法将检查过的灯泡转移给消费者。如果可以在超时时间过去之前将元素传输到使用者,则此方法将执行此操作。

避免使用transfer()方法,这可能会无限期地堵塞螺纹。

为了模拟生产者检查灯泡所花的时间,相应的线程将在 0 到 5 之间随机休眠几秒(5 秒是检查灯泡所需的最长时间)。如果消费者在此时间之后不可用,则会花费更多时间(在tryTransfer()中),直到消费者可用或超时结束。

另一方面,使用另一个类实现使用者,如下所示:

private static volatile boolean runningConsumer;
...
private static class Consumer implements Runnable {
  @Override
  public void run() {
    while (runningConsumer) {
      try {
        String bulb = queue.poll(
          MAX_PROD_TIME_MS, TimeUnit.MILLISECONDS);
        if (bulb != null) {
          Thread.sleep(rnd.nextInt(MAX_CONS_TIME_MS));
          logger.info(() -> "Packed: " + bulb);
        }
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        logger.severe(() -> "Exception: " + ex);
        break;
      }
    }
  }
}

消费者可以通过queue.take()方法从生产者处取灯泡。此方法检索并删除此队列的头,如有必要,请等待,直到灯泡可用。也可以调用poll()方法,在该方法中检索并移除队列的头,或者如果该队列为空,则返回null。但这两个都不适合我们。如果生产者不在,消费者可能仍然停留在take()方法中。另一方面,如果队列是空的(生产者现在正在检查当前灯泡),poll()方法将很快被一次又一次地调用,导致伪重复。解决这个问题的方法是poll(long timeout, TimeUnit unit)。此方法检索并删除此队列的头,并在指定的等待时间内(如果需要)等待灯泡变为可用。仅当等待时间过后队列为空时,才会返回null

为了模拟耗电元件包装灯泡所需的时间,相应的线程将在 0 到 7 之间随机休眠几秒(7 秒是包装灯泡所需的最长时间)。

启动生产者和消费者是一项非常简单的任务,它是通过一种名为startAssemblyLine()的方法完成的,如下所示:

public static void startAssemblyLine() {
  if (runningProducer || runningConsumer) {
    logger.info("Assembly line is already running ...");
    return;
  }
  logger.info("\n\nStarting assembly line ...");
  logger.info(() -> "Remaining bulbs from previous run: \n"
    + queue + "\n\n");
  runningProducer = true;
  producerService = Executors.newSingleThreadExecutor();
  producerService.execute(producer);
  runningConsumer = true;
  consumerService = Executors.newSingleThreadExecutor();
  consumerService.execute(consumer);
}

停止装配线是一个微妙的过程,可以通过不同的场景来解决。主要是,当装配线停止时,生产者应检查当前灯泡作为最后一个灯泡,消费者必须包装它。生产者可能需要等待消费者包装好当前灯泡,然后才能转移最后一个灯泡;此外,消费者也必须包装好这个灯泡。

为了遵循此场景,我们首先停止生产者,然后停止消费者:

public static void stopAssemblyLine() {
  logger.info("Stopping assembly line ...");
  boolean isProducerDown = shutdownProducer();
  boolean isConsumerDown = shutdownConsumer();
  if (!isProducerDown || !isConsumerDown) {
    logger.severe("Something abnormal happened during
      shutting down the assembling line!");
    System.exit(0);
  }
  logger.info("Assembling line was successfully stopped!");
}
private static boolean shutdownProducer() {
  runningProducer = false;
  return shutdownExecutor(producerService);
}
private static boolean shutdownConsumer() {
  runningConsumer = false;
  return shutdownExecutor(consumerService);
}

最后,我们给生产者和消费者足够的时间来正常停止(不中断线程)。这在shutdownExecutor()方法中发生,如下所示:

private static boolean shutdownExecutor(ExecutorService executor) {
  executor.shutdown();
  try {
    if (!executor.awaitTermination(TIMEOUT_MS * 2,
        TimeUnit.MILLISECONDS)) {
      executor.shutdownNow();
      return executor.awaitTermination(TIMEOUT_MS * 2,
        TimeUnit.MILLISECONDS);
    }
    return true;
  } catch (InterruptedException ex) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
    logger.severe(() -> "Exception: " + ex);
  }
  return false;
}

我们要做的第一件事是将runningProducer static变量设置为false。这将破坏while(runningProducer),因此这将是最后一次检查灯泡。此外,我们启动生产者的关闭程序。

对于消费者,我们要做的第一件事是将runningConsumer static变量设置为false。这将打破while(runningConsumer),因此这将是最后一个灯泡包装。此外,我们启动耗电元件的关闭程序。

让我们看看装配线的可能执行(运行 10 秒):

AssemblyLine.startAssemblyLine();
Thread.sleep(10 * 1000);
AssemblyLine.stopAssemblyLine();

可能的输出如下:

Starting assembly line ...
...
[2019-04-14 07:39:40] [INFO] Checked: bulb-89
[2019-04-14 07:39:43] [INFO] Packed: bulb-89
...
Stopping assembly line ...
...
[2019-04-14 07:39:53] [INFO] Packed: bulb-322
Assembling line was successfully stopped!

一般来说,如果停产需要很长时间(就好像停产一样),那么生产者和消费者的数量和/或生产和消费时间之间可能存在不平衡率。您可能需要添加或减去生产者或消费者。

生产者不等待消费者出现

如果生产者检查灯泡的速度比消费者包装灯泡的速度快,那么他们很可能会决定采用以下工作流程:

  • 生产者将逐一检查灯泡,并将其推入队列
  • 消费者将从队列中轮询并打包灯泡

由于消费者比生产者慢,队列将容纳已检查但未包装的灯泡(我们可以假设有空队列的可能性很低)。在下图中,我们有生产者、消费者和用于存储已检查但未包装灯泡的队列:

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZpLZp4de-1657345732717)(https://github.com/apachecn/apachecn-java-zh/raw/master/docs/java-coding-prob/img/00124b9f-15a5-4afd-993e-ee96b1ec47d1.png)]

为了形成这种情况,我们可以依赖于ConcurrentLinkedQueue(或LinkedBlockingQueue)。这是一个基于链接节点的无限线程安全队列:

private static final Queue<String> queue 
  = new ConcurrentLinkedQueue<>();

为了在队列中推一个灯泡,生产者调用offer()方法:

queue.offer(bulb);

另一方面,消费者使用poll()方法处理队列中的灯泡(因为消费者比生产者慢,所以当poll()返回null时应该是罕见的情况):

String bulb = queue.poll();

让我们第一次启动装配线 10 秒钟。这将输出以下内容:

Starting assembly line ...
...
[2019-04-14 07:44:58] [INFO] Checked: bulb-827
[2019-04-14 07:44:59] [INFO] Checked: bulb-257
[2019-04-14 07:44:59] [INFO] Packed: bulb-827
...
Stopping assembly line ...
...
[2019-04-14 07:45:08] [INFO] Checked: bulb-369
[2019-04-14 07:45:09] [INFO] Packed: bulb-690
...
Assembling line was successfully stopped!

此时,装配线停止,在队列中,我们有以下内容(这些灯泡已检查,但未包装):

[bulb-968, bulb-782, bulb-627, bulb-886, ...]

我们重新启动装配线并检查突出显示的行,这表明消费者从停止的位置恢复其工作:

Starting assembly line ...
[2019-04-14 07:45:12] [INFO ] Packed: bulb-968 [2019-04-14 07:45:12] [INFO ] Checked: bulb-812
[2019-04-14 07:45:12] [INFO ] Checked: bulb-470
[2019-04-14 07:45:14] [INFO ] Packed: bulb-782 [2019-04-14 07:45:15] [INFO ] Checked: bulb-601
[2019-04-14 07:45:16] [INFO ] Packed: bulb-627 ...

Java 编程问题:十、并发-线程池、可调用对象和同步器3https://developer.aliyun.com/article/1426163

相关文章
|
5天前
|
安全 Java 数据处理
Python网络编程基础(Socket编程)多线程/多进程服务器编程
【4月更文挑战第11天】在网络编程中,随着客户端数量的增加,服务器的处理能力成为了一个重要的考量因素。为了处理多个客户端的并发请求,我们通常需要采用多线程或多进程的方式。在本章中,我们将探讨多线程/多进程服务器编程的概念,并通过一个多线程服务器的示例来演示其实现。
|
4天前
|
安全 算法 Java
深入理解Java并发编程:线程安全与性能优化
【4月更文挑战第11天】 在Java中,高效的并发编程是提升应用性能和响应能力的关键。本文将探讨Java并发的核心概念,包括线程安全、锁机制、线程池以及并发集合等,同时提供实用的编程技巧和最佳实践,帮助开发者在保证线程安全的前提下,优化程序性能。我们将通过分析常见的并发问题,如竞态条件、死锁,以及如何利用现代Java并发工具来避免这些问题,从而构建更加健壮和高效的多线程应用程序。
|
17小时前
|
设计模式 运维 安全
深入理解Java并发编程:线程安全与性能优化
【4月更文挑战第15天】在Java开发中,多线程编程是提升应用程序性能和响应能力的关键手段。然而,它伴随着诸多挑战,尤其是在保证线程安全的同时如何避免性能瓶颈。本文将探讨Java并发编程的核心概念,包括同步机制、锁优化、线程池使用以及并发集合等,旨在为开发者提供实用的线程安全策略和性能优化技巧。通过实例分析和最佳实践的分享,我们的目标是帮助读者构建既高效又可靠的多线程应用。
|
1天前
|
SQL 安全 Java
Java安全编程:防范网络攻击与漏洞
【4月更文挑战第15天】本文强调了Java安全编程的重要性,包括提高系统安全性、降低维护成本和提升用户体验。针对网络攻击和漏洞,提出了防范措施:使用PreparedStatement防SQL注入,过滤和转义用户输入抵御XSS攻击,添加令牌对抗CSRF,限制文件上传类型和大小以防止恶意文件,避免原生序列化并确保数据完整性。及时更新和修复漏洞是关键。程序员应遵循安全编程规范,保障系统安全。
|
2天前
|
Java API 调度
安卓多线程和并发处理:提高应用效率
【4月更文挑战第13天】本文探讨了安卓应用中多线程和并发处理的优化方法,包括使用Thread、AsyncTask、Loader、IntentService、JobScheduler、WorkManager以及线程池。此外,还介绍了RxJava和Kotlin协程作为异步编程工具。理解并恰当运用这些技术能提升应用效率,避免UI卡顿,确保良好用户体验。随着安卓技术发展,更高级的异步处理工具将助力开发者构建高性能应用。
|
2天前
|
Java 编译器
Java并发编程中的锁优化策略
【4月更文挑战第13天】 在Java并发编程中,锁是一种常见的同步机制,用于保证多个线程之间的数据一致性。然而,不当的锁使用可能导致性能下降,甚至死锁。本文将探讨Java并发编程中的锁优化策略,包括锁粗化、锁消除、锁降级等方法,以提高程序的执行效率。
10 4
|
3天前
|
存储 Java 编译器
对象的交响曲:深入理解Java面向对象的绝妙之处
对象的交响曲:深入理解Java面向对象的绝妙之处
23 0
对象的交响曲:深入理解Java面向对象的绝妙之处
|
3天前
|
存储 安全 Java
Java语法掌握:打好编程基础的关键(二)
Java语法掌握:打好编程基础的关键
24 0
|
3天前
|
存储 Java
Java语法掌握:打好编程基础的关键(一)
Java语法掌握:打好编程基础的关键
9 0
Java语法掌握:打好编程基础的关键(一)
|
4天前
|
存储 Java 关系型数据库
掌握Java 8 Stream API的艺术:详解流式编程(一)
掌握Java 8 Stream API的艺术:详解流式编程
19 1