聊聊并发编程的 12 种业务场景
前言
并发编程是一项非常重要的技术,无论在面试,还是工作中出现的频率非常高。
并发编程说白了就是多线程编程,但多线程一定比单线程效率更高?
答:不一定,要看具体业务场景。
毕竟如果使用了多线程,那么线程之间的竞争和抢占 cpu 资源,线程的上下文切换,也是相对来说比较耗时的操作。
下面这几个问题在面试中,你必定遇到过:
你在哪来业务场景中使用过多线程?
怎么用的?
踩过哪些坑?
今天聊聊我之前在项目中用并发编程的 12 种业务场景,给有需要的朋友一个参考。
1. 简单定时任务
各位亲爱的朋友,你没看错,Thread 类真的能做定时任务。如果你看过一些定时任务框架的源码,你最后会发现,它们的底层也会使用 Thread 类。
实现这种定时任务的具体代码如下:
使用 Thread 类可以做最简单的定时任务,在 run 方法中有个 while 的死循环(当然还有其他方式),执行我们自己的任务。有个需要特别注意的地方是,需要用 try...catch 捕获异常,否则如果出现异常,就直接退出循环,下次将无法继续执行了。
但这种方式做的定时任务,只能周期性执行,不能支持定时在某个时间点执行。
特别提醒一下,该线程建议定义成守护线程,可以通过 setDaemon 方法设置,让它在后台默默执行就好。
使用场景:比如项目中有时需要每隔 5 分钟去下载某个文件,或者每隔 10 分钟去读取模板文件生成静态 html 页面等等,一些简单的周期性任务场景。
使用 Thread 类做定时任务的优缺点:
优点:这种定时任务非常简单,学习成本低,容易入手,对于那些简单的周期性任务,是个不错的选择。
缺点:不支持指定某个时间点执行任务,不支持延迟执行等操作,功能过于单一,无法应对一些较为复杂的场景。
2.监听器
有时候,我们需要写个监听器,去监听某些数据的变化。
比如:我们在使用 canal 的时候,需要监听 binlog 的变化,能够及时把数据库中的数据,同步到另外一个业务数据库中。
如果直接写一个监听器去监听数据就太没意思了,我们想实现这样一个功能:在配置中心有个开关,配置监听器是否开启,如果开启了使用单线程异步执行。
【文章福利】另外小编还整理了一些 C++后台开发面试题,教学视频,后端学习路线图免费分享,需要的可以自行添加:Q群:720209036 点击加入~ 群文件共享
小编强力推荐 C++后台开发免费学习地址:C/C++Linux服务器开发高级架构师/C++后台开发架构师
主要代码如下:
在 start 方法中开启了一个线程,在该线程中异步执行 handle 方法的具体任务。然后通过调用 stop 方法,可以停止该线程。
其中,使用 volatile 关键字控制的 running 变量作为开关,它可以控制线程中的状态。
接下来,有个比较关键的点是:如何通过配置中心的配置,控制这个开关呢?
以 apollo 配置为例,我们在配置中心的后台,修改配置之后,自动获取最新配置的核心代码如下:
通过 apollo 的 ApolloConfigChangeListener 注解,可以监听配置参数的变化。
如果 test.canal.enable 开关配置的 true,则调用 canalService 类的 start 方法开启 canal 数据同步功能。如果开关配置的 false,则调用 canalService 类的 stop 方法,自动停止 canal 数据同步功能。
3.收集日志
在某些高并发的场景中,我们需要收集部分用户的日志(比如:用户登录的日志),写到数据库中,以便于做分析。
但由于项目中,还没有引入消息中间件,比如:kafka、rocketmq 等。
如果直接将日志同步写入数据库,可能会影响接口性能。
所以,大家很自然想到了异步处理。
实现这个需求最简单的做法是,开启一个线程,异步写入数据到数据库即可。
这样做,可以是可以。
但如果用户登录操作的耗时,比异步写入数据库的时间要少得多。这样导致的结果是:生产日志的速度,比消费日志的速度要快得多,最终的性能瓶颈在消费端。
其实,还有更优雅的处理方式,虽说没有使用消息中间件,但借用了它的思想。
这套记录登录日志的功能,分为:日志生产端、日志存储端和日志消费端。
如下图所示:
先定义了一个阻塞队列。
然后定义了一个日志的生产者。
接下来,定义了日志的消费者。
当然,这个例子中使用单线程接收登录日志,为了提升性能,也可以使用线程池来处理业务逻辑(比如:写入数据库)等。
4.excel 导入
我们可能会经常收到运营同学提过来的 excel 数据导入需求,比如:将某一大类下的所有子类一次性导入系统,或者导入一批新的供应商数据等等。
我们以导入供应商数据为例,它所涉及的业务流程很长,比如:
调用天眼查接口校验企业名称和统一社会信用代码。
写入供应商基本表
写入组织表
给供应商自动创建一个用户
给该用户分配权限
自定义域名
发站内通知
等等。
如果在程序中,解析完 excel,读取了所有数据之后。用单线程一条条处理业务逻辑,可能耗时会非常长。
为了提升 excel 数据导入效率,非常有必要使用多线程来处理。
当然在 java 中实现多线程的手段有很多种,下面重点聊聊 java8 中最简单的实现方式:parallelStream。
伪代码如下:
parallelStream 是一个并行执行的流,它默认通过 ForkJoinPool 实现的,能提高你的多线程任务的速度。
ForkJoinPool 处理的过程会分而治之,它的核心思想是:将一个大任务切分成多个小任务。每个小任务都能单独执行,最后它会把所用任务的执行结果进行汇总。
下面用一张图简单介绍一下 ForkJoinPool 的原理:
当然除了 excel 导入之外,还有类似的读取文本文件,也可以用类似的方法处理。
温馨的提醒一下,如果一次性导入的数据非常多,用多线程处理,可能会使系统的 cpu 使用率飙升,需要特别关注。
5.查询接口
很多时候,我们需要在某个查询接口中,调用其他服务的接口,组合数据之后,一起返回。
比如有这样的业务场景:
在用户信息查询接口中需要返回:用户名称、性别、等级、头像、积分、成长值等信息。
而用户名称、性别、等级、头像在用户服务中,积分在积分服务中,成长值在成长值服务中。为了汇总这些数据统一返回,需要另外提供一个对外接口服务。
于是,用户信息查询接口需要调用用户查询接口、积分查询接口 和 成长值查询接口,然后汇总数据统一返回。
调用过程如下图所示:
调用远程接口总耗时 530ms = 200ms + 150ms + 180ms
显然这种串行调用远程接口性能是非常不好的,调用远程接口总的耗时为所有的远程接口耗时之和。
那么如何优化远程接口性能呢?
既然串行调用多个远程接口性能很差,为什么不改成并行呢?
如下图所示:
调用远程接口总耗时 200ms = 200ms(即耗时最长的那次远程接口调用)
在 java8 之前可以通过实现 Callable 接口,获取线程返回结果。
java8 以后通过 CompleteFuture 类实现该功能。我们这里以 CompleteFuture 为例:
温馨提醒一下,这两种方式别忘了使用线程池。示例中我用到了 executor,表示自定义的线程池,为了防止高并发场景下,出现线程过多的问题。
6.获取用户上下文
不知道你在项目开发时,有没有遇到过这样的需求:用户登录之后,在所有的请求接口中,通过某个公共方法,就能获取到当前登录用户的信息?
获取的用户上下文,我们以 CurrentUser 为例。
CurrentUser 内部包含了一个 ThreadLocal 对象,它负责保存当前线程的用户上下文信息。当然为了保证在线程池中,也能从用户上下文中获取到正确的用户信息,这里用了阿里的 TransmittableThreadLocal。伪代码如下:
这里为什么用了阿里的 TransmittableThreadLocal,而不是普通的 ThreadLocal 呢?在线程池中,由于线程会被多次复用,导致从普通的 ThreadLocal 中无法获取正确的用户信息。父线程中的参数,没法传递给子线程,而 TransmittableThreadLocal 很好解决了这个问题。
然后在项目中定义一个全局的 spring mvc 拦截器,专门设置用户上下文到 ThreadLocal 中。伪代码如下:
用户在请求我们接口时,会先触发该拦截器,它会根据用户 cookie 中的 token,调用调用接口获取 redis 中的用户信息。如果能获取到,说明用户已经登录,则把用户信息设置到 CurrentUser 类的 ThreadLocal 中。
接下来,在 api 服务的下层,即 business 层的方法中,就能轻松通过 CurrentUser.getCurrent();方法获取到想要的用户上下文信息了。
这套用户体系的想法是很 good 的,但深入使用后,发现了一个小插曲:
api 服务和 mq 消费者服务都引用了 business 层,business 层中的方法两个服务都能直接调用。
我们都知道在 api 服务中用户是需要登录的,而 mq 消费者服务则不需要登录。
如果 business 中的某个方法刚开始是给 api 开发的,在方法深处使用了 CurrentUser.getCurrent();获取用户上下文。但后来,某位新来的帅哥在 mq 消费者中也调用了那个方法,并未发觉这个小机关,就会中招,出现找不到用户上下文的问题。
所以我当时的第一个想法是:代码没做兼容处理,因为之前这类问题偶尔会发生一次。
想要解决这个问题,其实也很简单。只需先判断一下能否从 CurrentUser 中获取用户信息,如果不能,则取配置的系统用户信息。伪代码如下:
这种简单无公害的代码,如果只是在一两个地方加还 OK。
此外,众所周知,SimpleDateFormat 在 java8 以前,是用来处理时间的工具类,它是非线程安全的。也就是说,用该方法解析日期会有线程安全问题。
为了避免线程安全问题的出现,我们可以把 SimpleDateFormat 对象定义成局部变量。但如果你一定要把它定义成静态变量,可以使用 ThreadLocal 保存日期,也能解决线程安全问题。
8. 传递参数
之前见过有些同事写代码时,一个非常有趣的用法,即:使用 MDC 传递参数。
MDC 是什么?
MDC 是 org.slf4j 包下的一个类,它的全称是 Mapped Diagnostic Context,我们可以认为它是一个线程安全的存放诊断日志的容器。
MDC 的底层是用了 ThreadLocal 来保存数据的。
例如现在有这样一种场景:我们使用 RestTemplate 调用远程接口时,有时需要在 header 中传递信息,比如:traceId,source 等,便于在查询日志时能够串联一次完整的请求链路,快速定位问题。
这种业务场景就能通过 ClientHttpRequestInterceptor 接口实现,具体做法如下:
第一步,定义一个 LogFilter 拦截所有接口请求,在 MDC 中设置 traceId:
第二步,实现 ClientHttpRequestInterceptor 接口,MDC 中获取当前请求的 traceId,然后设置到 header 中:
第三步,定义配置类,配置上面定义的 RestTemplateInterceptor 类:
其中 MdcUtil 其实是利用 MDC 工具在 ThreadLocal 中存储和获取 traceId
当然,这个例子中没有演示 MdcUtil 类的 add 方法具体调的地方,我们可以在 filter 中执行接口方法之前,生成 traceId,调用 MdcUtil 类的 add 方法添加到 MDC 中,然后在同一个请求的其他地方就能通过 MdcUtil 类的 get 方法获取到该 traceId。
能使用 MDC 保存 traceId 等参数的根本原因是,用户请求到应用服务器,Tomcat 会从线程池中分配一个线程去处理该请求。
那么该请求的整个过程中,保存到 MDC 的 ThreadLocal 中的参数,也是该线程独享的,所以不会有线程安全问题。
9. 模拟高并发
有时候我们写的接口,在低并发的场景下,一点问题都没有。
但如果一旦出现高并发调用,该接口可能会出现一些意想不到的问题。
为了防止类似的事情发生,一般在项目上线前,我们非常有必要对接口做一下压力测试。
当然,现在已经有比较成熟的压力测试工具,比如:Jmeter、LoadRunner 等。
如果你觉得下载压测工具比较麻烦,也可以手写一个简单的模拟并发操作的工具,用 CountDownLatch 就能实现,例如:
10. 处理 mq 消息
在高并发的场景中,消息积压问题,可以说如影随形,真的没办法从根本上解决。表面上看,已经解决了,但后面不知道什么时候,就会冒出一次,比如这次:
有天下午,产品过来说:有几个商户投诉过来了,他们说菜品有延迟,快查一下原因。
这次问题出现得有点奇怪。
为什么这么说?
首先这个时间点就有点奇怪,平常出问题,不都是中午或者晚上用餐高峰期吗?怎么这次问题出现在下午?
根据以往积累的经验,我直接看了 kafka 的 topic 的数据,果然上面消息有积压,但这次每个 partition 都积压了十几万的消息没有消费,比以往加压的消息数量增加了几百倍。这次消息积压得极不寻常。
我赶紧查服务监控看看消费者挂了没,还好没挂。又查服务日志没有发现异常。这时我有点迷茫,碰运气问了问订单组下午发生了什么事情没?他们说下午有个促销活动,跑了一个 JOB 批量更新过有些商户的订单信息。
这时,我一下子如梦初醒,是他们在 JOB 中批量发消息导致的问题。怎么没有通知我们呢?实在太坑了。
虽说知道问题的原因了,倒是眼前积压的这十几万的消息该如何处理呢?
此时,如果直接调大 partition 数量是不行的,历史消息已经存储到 4 个固定的 partition,只有新增的消息才会到新的 partition。我们重点需要处理的是已有的 partition。
直接加服务节点也不行,因为 kafka 允许同组的多个 partition 被一个 consumer 消费,但不允许一个 partition 被同组的多个 consumer 消费,可能会造成资源浪费。
看来只有用多线程处理了。
为了紧急解决问题,我改成了用线程池处理消息,核心线程和最大线程数都配置成了 50。
大致用法如下:
先定义一个线程池:
再定义一个消息的 consumer:
在定义的 Runable 实现类中处理业务逻辑:
果然,调整之后消息积压数量确实下降的非常快,大约半小时后,积压的消息就非常顺利的处理完了。
但此时有个更严重的问题出现:我收到了报警邮件,有两个订单系统的节点 down 机了。。。
11. 统计数量
在多线程的场景中,有时候需要统计数量,比如:用多线程导入供应商数据时,统计导入成功的供应商数有多少。
如果这时候用 count++统计次数,最终的结果可能会不准。因为 count++并非原子操作,如果多个线程同时执行该操作,则统计的次数,可能会出现异常。
为了解决这个问题,就需要使用 concurent 的 atomic 包下面的类,比如:AtomicInteger、AtomicLong 等。
AtomicInteger 的底层说白了使用自旋锁+CAS。
自旋锁说白了就是一个死循环。
而 CAS 是比较和交换的意思。
它的实现逻辑是:将内存位置处的旧值与预期值进行比较,若相等,则将内存位置处的值替换为新值。若不相等,则不做任何操作。
12. 延迟定时任务
我们经常有延迟处理数据的需求,比如:如果用户下单后,超过 30 分钟还未完成支付,则系统自动将该订单取消。
这里需求就可以使用延迟定时任务实现。
ScheduledExecutorService 是 JDK1.5+版本引进的定时任务,该类位于 java.util.concurrent 并发包下。
ScheduledExecutorService 是基于多线程的,设计的初衷是为了解决 Timer 单线程执行,多个任务之间会互相影响的问题。
它主要包含 4 个方法:
schedule(Runnable command,long delay,TimeUnit unit),带延迟时间的调度,只执行一次,调度之后可通过 Future.get()阻塞直至任务执行完毕。
schedule(Callablecallable,long delay,TimeUnit unit),带延迟时间的调度,只执行一次,调度之后可通过 Future.get()阻塞直至任务执行完毕,并且可以获取执行结果。
scheduleAtFixedRate,表示以固定频率执行的任务,如果当前任务耗时较多,超过定时周期 period,则当前任务结束后会立即执行。
scheduleWithFixedDelay,表示以固定延时执行任务,延时是相对当前任务结束为起点计算开始时间。
实现这种定时任务的具体代码如下:
调用 ScheduledExecutorService 类的 scheduleAtFixedRate 方法实现周期性任务,每隔 1 秒钟执行一次,每次延迟 1 秒再执行。
这种定时任务是阿里巴巴开发者规范中用来替代 Timer 类的方案,对于多线程执行周期性任务,是个不错的选择。
使用 ScheduledExecutorService 类做延迟定时任务的优缺点:
优点:基于多线程的定时任务,多个任务之间不会相关影响,支持周期性的执行任务,并且带延迟功能。
缺点:不支持一些较复杂的定时规则。
当然,你也可以使用分布式定时任务,比如:xxl-job 或者 elastic-job 等等。
其实,在实际工作中我使用多线程的场景远远不只这 12 种,在这里只是抛砖引玉,介绍了一些我认为比较常见的业务场景。
参考资料
推荐一个零声教育 C/C++后台开发的免费公开课程,个人觉得老师讲得不错,分享给大家:C/C++后台开发高级架构师,内容包括Linux,Nginx,ZeroMQ,MySQL,Redis,fastdfs,MongoDB,ZK,流媒体,CDN,P2P,K8S,Docker,TCP/IP,协程,DPDK等技术内容,立即学习
评论