【并发设计模式】聊聊Thread-Per-Message与Worker-Thread模式

2024-01-02 07:52

本文主要是介绍【并发设计模式】聊聊Thread-Per-Message与Worker-Thread模式,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

在并发编程中,核心就是同步、互斥、分工。

同步是多个线程之间按照一定的顺序进行执行,比如A执行完,B在执行。而互斥是多个线程之间对于共享资源的互斥。两个侧重点不一样,同步关注的是执行顺序,互斥关注的是资源的排他性访问。

分工则是从一个宏观的角度,将任务庖丁解牛,将一个大任务进行拆分,每个线程执行一个任务的子集。主要的设计模式就是Thread-Per-Message(来一个任务,新建一个线程执行)、Worker-Thread(复用线程池)、生产者 - 消费者模式。本篇我们先介绍前两个。

  • Thread-Per-Message 模式需要注意线程的创建,销毁以及是否会导致OOM。
  • Worker Thread 模式需要注意死锁问题,提交的任务之间不要有依赖性。
  • 生产者 - 消费者模式可以直接使用线程池来实现

生活场景触发

我们来想一个实际的参观厨师、服务员的方式。如果来一个客人,再去招聘人,然后开始做饭,显然销量不高,所以就会提前雇佣好一批人,来了客人直接做饭。但是显然也有饭店火爆的情况,那么就让客人先在休息区等待,等待进行排号吃饭。因为目前饭店已经达到的上限。对比其实就是上述的三种方式。

Thread-Per-Message

这种方式比较好理解,针对于每个客户端的请求,来一个请求就新建一个Thread进行处理。
但是显而易见,这种方式新建线程、销毁线程的操作是很耗时,比较浪费资源。并且如果大量的线程处理任务耗时比较久,那么就会出现OOM,所以JUC中就提供了线程中方式,根据需要配置线程池进行处理任务。
在GO语言中有更加轻量级的协程,以及java中Loom 推荐你可以看看。

解决方案:短期可以增大JVM内存配置,调整大新生代大小,长期解决NIO或者AIO等

Loom

Project Loom is to intended to explore, incubate and deliver Java VM features and APIs built on top of them for the purpose of supporting easy-to-use, high-throughput lightweight concurrency and new programming models on the Java platform.
This OpenJDK project is sponsored by the HotSpot Group.

Code


/*** @author qxlx* @date 2023/12/31 10:33 PM*/
public class ServerTest {public static void main(String[] args) throws IOException {final ServerSocketChannel ssc = ServerSocketChannel.open().bind(new InetSocketAddress(8088));while (true) {SocketChannel sc = ssc.accept();new Thread(() -> {ByteBuffer rb = ByteBuffer.allocate(1024);try {sc.read(rb);TimeUnit.SECONDS.sleep(2000);System.out.println(Thread.currentThread().getName() + " 接收的数据:" + rb.toString());ByteBuffer wb = (ByteBuffer) rb.flip();sc.write(wb);sc.close();} catch (Exception e) {e.printStackTrace();}}, "Thread-" + Math.random()).start();}}}

Worker-Thread

Woker-Thread 其实就是我提前雇佣一批工人,等待干活,来活就干,没活就休息。可以避免频繁的创建线程。
在这里插入图片描述
Worker Thread 模式能避免线程频繁创建、销毁的问题,而且能够限制线程的最大数量。
Java 语言里可以直接使用线程池来实现 Worker Thread 模式,线程池是一个非常基础和优秀 的工具类,甚至有些大厂的编码规范都不允许用 new Thread() 来创建线程,必须使用线程池。

Code

public static void main(String[] args) throws IOException {ExecutorService threadPoolExecutor = new ThreadPoolExecutor(10,20,30000,TimeUnit.SECONDS,new ArrayBlockingQueue<>(10));final ServerSocketChannel ssc = ServerSocketChannel.open().bind(new InetSocketAddress(8088));while (true) {SocketChannel sc = ssc.accept();threadPoolExecutor.execute(() -> {ByteBuffer rb = ByteBuffer.allocate(1024);try {sc.read(rb);TimeUnit.SECONDS.sleep(2000);System.out.println(Thread.currentThread().getName() + " 接收的数据:" + rb.toString());ByteBuffer wb = (ByteBuffer) rb.flip();sc.write(wb);sc.close();} catch (Exception e) {e.printStackTrace();}});}}

避免死锁

在实际的开发中,使用线程池需要注意任务之间是否有依赖关系,否则有以来关系的话,可能会引起线程死锁。

在这里插入图片描述
如下就是2个线程,执行的时候,因为线程池线程使用完毕,本来需要4个,但是只有两个,另外两个线程任务执行不了,所以就死锁了。

package com.jia.dp;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/*** @author qxlx* @date 2024/1/1 11:17 AM*/
public class ThreadPoolDeadLockTest {public static void main(String[] args) throws InterruptedException {ExecutorService threadPool = Executors.newFixedThreadPool(2);CountDownLatch l1 = new CountDownLatch(2);System.out.println("l1-begin");// 大任务l1 2个for (int i = 0; i < 2; i++) {threadPool.execute(()-> {CountDownLatch l2 = new CountDownLatch(2);System.out.println("l2-begin");//小任务l2 2个for (int j = 0; j < 2; j++) {threadPool.execute(()->{l2.countDown();});}try {l2.await();l1.countDown();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("l2-end");});}System.out.println("l1-end");l1.await();}}
"Reference Handler" #2 daemon prio=10 os_prio=31 tid=0x00007f8d4480a800 nid=0x3223 in Object.wait() [0x000000030646a000]java.lang.Thread.State: WAITING (on object monitor)at java.lang.Object.wait(Native Method)- waiting on <0x0000000715586bf8> (a java.lang.ref.Reference$Lock)at java.lang.Object.wait(Object.java:502)at java.lang.ref.Reference.tryHandlePending(Reference.java:191)- locked <0x0000000715586bf8> (a java.lang.ref.Reference$Lock)at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:153)"main" #1 prio=5 os_prio=31 tid=0x00007f8d56010800 nid=0x1903 waiting on condition [0x0000000305949000]java.lang.Thread.State: WAITING (parking)at sun.misc.Unsafe.park(Native Method)- parking to wait for  <0x00000007156ac720> (a java.util.concurrent.CountDownLatch$Sync)at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)at com.jia.dp.ThreadPoolDeadLockTest.main(ThreadPoolDeadLockTest.java:40)

在这里插入图片描述
可以发现是因为l1.await() 阻塞,l1阻塞的原因就是l2线程任务没有执行完毕,l2线程任务没有线程资源可以处理任务,所以就是死锁了。

解决方案
1.调整线程池的大小,更加方便的是,使用不同的线程池任务进行处理不同的任务。

总结

本篇介绍了两种分工协作的方式,一种是来一个任务new一个线程处理,另外一种就是通过线程池进行达到线程的复用。实际生产中是采用后者的方式。

这篇关于【并发设计模式】聊聊Thread-Per-Message与Worker-Thread模式的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/561813

相关文章

Java并发编程必备之Synchronized关键字深入解析

《Java并发编程必备之Synchronized关键字深入解析》本文我们深入探索了Java中的Synchronized关键字,包括其互斥性和可重入性的特性,文章详细介绍了Synchronized的三种... 目录一、前言二、Synchronized关键字2.1 Synchronized的特性1. 互斥2.

SpringBoot如何通过Map实现策略模式

《SpringBoot如何通过Map实现策略模式》策略模式是一种行为设计模式,它允许在运行时选择算法的行为,在Spring框架中,我们可以利用@Resource注解和Map集合来优雅地实现策略模式,这... 目录前言底层机制解析Spring的集合类型自动装配@Resource注解的行为实现原理使用直接使用M

Python异步编程中asyncio.gather的并发控制详解

《Python异步编程中asyncio.gather的并发控制详解》在Python异步编程生态中,asyncio.gather是并发任务调度的核心工具,本文将通过实际场景和代码示例,展示如何结合信号量... 目录一、asyncio.gather的原始行为解析二、信号量控制法:给并发装上"节流阀"三、进阶控制

Redis中高并发读写性能的深度解析与优化

《Redis中高并发读写性能的深度解析与优化》Redis作为一款高性能的内存数据库,广泛应用于缓存、消息队列、实时统计等场景,本文将深入探讨Redis的读写并发能力,感兴趣的小伙伴可以了解下... 目录引言一、Redis 并发能力概述1.1 Redis 的读写性能1.2 影响 Redis 并发能力的因素二、

C#原型模式之如何通过克隆对象来优化创建过程

《C#原型模式之如何通过克隆对象来优化创建过程》原型模式是一种创建型设计模式,通过克隆现有对象来创建新对象,避免重复的创建成本和复杂的初始化过程,它适用于对象创建过程复杂、需要大量相似对象或避免重复初... 目录什么是原型模式?原型模式的工作原理C#中如何实现原型模式?1. 定义原型接口2. 实现原型接口3

大数据spark3.5安装部署之local模式详解

《大数据spark3.5安装部署之local模式详解》本文介绍了如何在本地模式下安装和配置Spark,并展示了如何使用SparkShell进行基本的数据处理操作,同时,还介绍了如何通过Spark-su... 目录下载上传解压配置jdk解压配置环境变量启动查看交互操作命令行提交应用spark,一个数据处理框架

Nginx实现高并发的项目实践

《Nginx实现高并发的项目实践》本文主要介绍了Nginx实现高并发的项目实践,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录使用最新稳定版本的Nginx合理配置工作进程(workers)配置工作进程连接数(worker_co

Java实现状态模式的示例代码

《Java实现状态模式的示例代码》状态模式是一种行为型设计模式,允许对象根据其内部状态改变行为,本文主要介绍了Java实现状态模式的示例代码,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来... 目录一、简介1、定义2、状态模式的结构二、Java实现案例1、电灯开关状态案例2、番茄工作法状态案例

在JS中的设计模式的单例模式、策略模式、代理模式、原型模式浅讲

1. 单例模式(Singleton Pattern) 确保一个类只有一个实例,并提供一个全局访问点。 示例代码: class Singleton {constructor() {if (Singleton.instance) {return Singleton.instance;}Singleton.instance = this;this.data = [];}addData(value)

高并发环境中保持幂等性

在高并发环境中保持幂等性是一项重要的挑战。幂等性指的是无论操作执行多少次,其效果都是相同的。确保操作的幂等性可以避免重复执行带来的副作用。以下是一些保持幂等性的常用方法: 唯一标识符: 请求唯一标识:在每次请求中引入唯一标识符(如 UUID 或者生成的唯一 ID),在处理请求时,系统可以检查这个标识符是否已经处理过,如果是,则忽略重复请求。幂等键(Idempotency Key):客户端在每次