并发编程的12种业务场景

打印 上一主题 下一主题

主题 783|帖子 783|积分 2349

媒介

并发编程是一项非常紧张的技术,无论是在面试还是工作中都频繁出现。
并发编程说白了就是多线程编程,但是多线程肯定比单线程效率更高吗?
答案是不肯定,要看详细的业务场景。
究竟,假如使用了多线程,线程之间的竞争和抢占 CPU 资源、线程的上下文切换等操作相对来说比力耗时。
在面试中,你肯定会遇到以下几个标题:

  • 你在哪个业务场景中使用过多线程?
  • 如何使用多线程?
  • 踩过哪些坑?
今天聊聊我之前在项目中用并发编程的12种业务场景,给有需要的朋友一个参考。

1. 简朴定时任务

各位酷爱的朋友们,你没有看错,Thread类确实可以用来实现定时任务。假如你看过一些定时任务框架的源码,你最后会发现,它们的底层也会使用Thread类。
实现这种定时任务的详细代码如下:
  1. public static void init() {
  2.     new Thread(() -> {
  3.         while (true) {
  4.             try {
  5.                 System.out.println("下载文件");
  6.                 Thread.sleep(1000 * 60 * 5);
  7.             } catch (Exception e) {
  8.                 log.error(e);
  9.             }
  10.         }
  11.     }).start();
  12. }
复制代码
使用Thread类可以实现最简朴的定时任务。在run方法中,我们可以使用while死循环(固然另有其他方式)实行自己的任务。需要特别注意的是,需要使用try...catch捕捉异常,否则假如出现异常,就会直接退出循环,下次将无法继承实行。
但是,这种方式只能周期性实行,无法支持在指定时间点实行。
特别提示,建议将该线程界说为守护线程,可以通过setDaemon方法设置,让它在后台冷静实行就好。
使用场景:例如,项目中有时需要每隔5分钟下载某个文件,或者每隔10分钟读取模板文件天生静态html页面等。对于一些简朴的周期性任务场景,这种方式是不错的选择。
使用Thread类做定时任务的优缺点:


  • 优点:这种定时任务非常简朴,学习成本低,容易入手,对于那些简朴的周期性任务,是个不错的选择。
  • 缺点:不支持指定某个时间点实行任务,不支持延迟实行等操作,功能过于单一,无法应对一些较为复杂的场景。
2.监听器

有时候,我们需要编写一个监听器来监听某些数据的变化。
例如:在使用 canal 时,需要监听 binlog 的变化,以便将数据库中的数据及时同步到另一个业务数据库中。
假如直接编写一个监听器来监听数据,那就太无聊了。我们想要实现这样一个功能:在设置中心中有一个开关,用于控制监听器是否开启。假如开启了,则使用单线程异步实行。
主要代码如下:
  1. @Service
  2. public CanalService {
  3.     private volatile boolean running = false;
  4.     private Thread thread;
  5.     @Autowired
  6.     private CanalConnector canalConnector;
  7.     public void handle() {
  8. //连接canal
  9.         while(running) {
  10. //业务处理
  11.         }
  12.     }
  13.     public void start() {
  14.        thread = new Thread(this::handle, "name");
  15.        running = true;
  16.        thread.start();
  17.     }
  18.     public void stop() {
  19.        if(!running) {
  20.           return;
  21.        }
  22.        running = false;
  23.     }
  24. }
复制代码
在start方法中开启了一个线程,在该线程中异步实行handle方法的详细任务。然后通过调用stop方法,可以停止该线程。
其中,使用volatile关键字控制的running变量作为开关,它可以控制线程中的状态。
接下来,有个比力关键的点是:如何通过设置中心的设置,控制这个开关呢?
以apollo设置为例,我们在设置中心的后台,修改设置之后,主动获取最新设置的焦点代码如下:
  1. public class CanalConfig {
  2.     @Autowired
  3.     private CanalService canalService;
  4.     @ApolloConfigChangeListener
  5.     public void change(ConfigChangeEvent event) {
  6.         String value = event.getChange("test.canal.enable").getNewValue();
  7.         if(BooleanUtils.toBoolean(value)) {
  8.             canalService.start();
  9.         } else {
  10.             canalService.stop();
  11.         }
  12.     }
  13. }
复制代码
通过apollo的ApolloConfigChangeListener注解,可以监听设置参数的变化。
假如test.canal.enable开关设置的true,则调用canalService类的start方法开启canal数据同步功能。假如开关设置的false,则调用canalService类的stop方法,主动停止canal数据同步功能。
3.收集日志

在某些高并发的场景中,我们需要收集部门用户的日志(好比:用户登录的日志),写到数据库中,以便于做分析。
但由于项目中,还没有引入消息中间件,好比:kafka、rocketmq等。
假如直接将日志同步写入数据库,可能会影响接口性能。
以是,大家很天然想到了异步处理处罚。
实现这个需求最简朴的做法是,开启一个线程,异步写入数据到数据库即可。
这样做,可以是可以。
但假如用户登录操作的耗时,比异步写入数据库的时间要少得多。这样导致的结果是:生产日志的速度,比消费日志的速度要快得多,终极的性能瓶颈在消费端。
其实,另有更优雅的处理处罚方式,虽说没有使用消息中间件,但借用了它的思想。
这套记载登录日志的功能,分为:日志生产端、日志存储端和日志消费端。
如下图所示:

先界说了一个壅闭队列。
  1. @Component
  2. public class LoginLogQueue {
  3.     private static final int QUEUE_MAX_SIZE    = 1000;
  4.     private BlockingQueueblockingQueue queue = new LinkedBlockingQueue<>(QUEUE_MAX_SIZE);
  5. //生成消息
  6.     public boolean push(LoginLog loginLog) {
  7.         return this.queue.add(loginLog);
  8.     }
  9. //消费消息
  10.     public LoginLog poll() {
  11.         LoginLog loginLog = null;
  12.         try {
  13.             loginLog = this.queue.take();
  14.         } catch (InterruptedException e) {
  15.             e.printStackTrace();
  16.         }
  17.         return result;
  18.     }
  19. }
复制代码
然后界说了一个日志的生产者。
  1. @Service
  2. public class LoginSerivce {
  3.     @Autowired
  4.     private LoginLogQueue loginLogQueue;
  5.     public int login(UserInfo userInfo) {
  6. //业务处理
  7.         LoginLog loginLog = convert(userInfo);
  8.         loginLogQueue.push(loginLog);
  9.     }
  10. }
复制代码
接下来,界说了日志的消费者。
  1. @Service
  2. public class LoginInfoConsumer {
  3.     @Autowired
  4.     private LoginLogQueue queue;
  5.     @PostConstruct
  6.     public voit init {
  7.        new Thread(() -> {
  8.           while (true) {
  9.               LoginLog loginLog = queue.take();
  10. //写入数据库
  11.           }
  12.         }).start();
  13.     }
  14. }
复制代码
固然,这个例子中使用单线程接收登录日志,为了提拔性能,也可以使用线程池来处理处罚业务逻辑(好比:写入数据库)等。
4.excel导入

我们可能会经常收到运营同学提过来的excel数据导入需求,好比:将某一大类下的所有子类一次性导入体系,或者导入一批新的供应商数据等等。
我们以导入供应商数据为例,它所涉及的业务流程很长,好比:

  • 调用天眼查接口校验企业名称和统一社会名誉代码。
  • 写入供应商基本表
  • 写入构造表
  • 给供应商主动创建一个用户
  • 给该用户分配权限
  • 自界说域名
  • 发站内通知
等等。
假如在步伐中,分析完excel,读取了所有数据之后。用单线程一条条处理处罚业务逻辑,可能耗时会非常长。
为了提拔excel数据导入效率,非常有须要使用多线程来处理处罚。
固然在java中实现多线程的手段有许多种,下面重点聊聊java8中最简朴的实现方式:parallelStream。
伪代码如下:
  1. supplierList.parallelStream().forEach(x -> importSupplier(x));
复制代码
parallelStream是一个并行实行的流,它默认通过ForkJoinPool实现的,能进步你的多线程任务的速度。
ForkJoinPool处理处罚的过程会分而治之,它的焦点思想是:将一个大任务切分成多个小任务。每个小任务都能单独实行,最后它会把所用任务的实行结果进行汇总。
下面用一张图简朴介绍一下ForkJoinPool的原理:

固然除了excel导入之外,另有雷同的读取文本文件,也可以用雷同的方法处理处罚。
   温馨的提示一下,假如一次性导入的数据非常多,用多线程处理处罚,可能会使体系的cpu使用率飙升,需要特别关注。
  5.查询接口

许多时候,我们需要在某个查询接口中,调用其他服务的接口,组合数据之后,一起返回。
好比有这样的业务场景:
在用户信息查询接口中需要返回:用户名称、性别、品级、头像、积分、成长值等信息。
而用户名称、性别、品级、头像在用户服务中,积分在积分服务中,成长值在成长值服务中。为了汇总这些数据统一返回,需要另外提供一个对外接口服务。
于是,用户信息查询接口需要调用用户查询接口、积分查询接口 和 成长值查询接口,然后汇总数据统一返回。
调用过程如下图所示:

调用远程接口总耗时 530ms = 200ms + 150ms + 180ms
显然这种串行调用远程接口性能是非常不好的,调用远程接口总的耗时为所有的远程接口耗时之和。
那么如何优化远程接口性能呢?
既然串行调用多个远程接口性能很差,为什么不改成并行呢?
如下图所示:

调用远程接口总耗时 200ms = 200ms(即耗时最长的那次远程接口调用)
在java8之前可以通过实现Callable接口,获取线程返回结果。
java8以后通过CompleteFuture类实现该功能。我们这里以CompleteFuture为例:
  1. public UserInfo getUserInfo(Long id) throws InterruptedException, ExecutionException {
  2.     final UserInfo userInfo = new UserInfo();
  3.     CompletableFuture userFuture = CompletableFuture.supplyAsync(() -> {
  4.         getRemoteUserAndFill(id, userInfo);
  5.         return Boolean.TRUE;
  6.     }, executor);
  7.     CompletableFuture bonusFuture = CompletableFuture.supplyAsync(() -> {
  8.         getRemoteBonusAndFill(id, userInfo);
  9.         return Boolean.TRUE;
  10.     }, executor);
  11.     CompletableFuture growthFuture = CompletableFuture.supplyAsync(() -> {
  12.         getRemoteGrowthAndFill(id, userInfo);
  13.         return Boolean.TRUE;
  14.     }, executor);
  15.     CompletableFuture.allOf(userFuture, bonusFuture, growthFuture).join();
  16.     userFuture.get();
  17.     bonusFuture.get();
  18.     growthFuture.get();
  19.     return userInfo;
  20. }
复制代码
温馨提示一下,这两种方式别忘了使用线程池。示例中我用到了executor,表示自界说的线程池,为了防止高并发场景下,出现线程过多的标题。
6.获取用户上下文

不知道你在项目开发时,有没有遇到过这样的需求:用户登录之后,在所有的请求接口中,通过某个公共方法,就能获取到当前登录用户的信息?
获取的用户上下文,我们以CurrentUser为例。
CurrentUser内部包罗了一个ThreadLocal对象,它负责保存当火线程的用户上下文信息。固然为了保证在线程池中,也能从用户上下文中获取到正确的用户信息,这里用了阿里的TransmittableThreadLocal。伪代码如下:
  1. @Data
  2. public class CurrentUser {
  3.     private static final TransmittableThreadLocal<CurrentUser> THREA_LOCAL = new TransmittableThreadLocal<>();
  4.     private String id;
  5.     private String userName;
  6.     private String password;
  7.     private String phone;
  8.     ...
  9.     public statis void set(CurrentUser user) {
  10.       THREA_LOCAL.set(user);
  11.     }
  12.     public static void getCurrent() {
  13.       return THREA_LOCAL.get();
  14.     }
  15. }
复制代码
这里为什么用了阿里的TransmittableThreadLocal,而不是平凡的ThreadLocal呢?在线程池中,由于线程会被多次复用,导致从平凡的ThreadLocal中无法获取正确的用户信息。父线程中的参数,没法传递给子线程,而TransmittableThreadLocal很好解决了这个标题。
然后在项目中界说一个全局的spring mvc拦截器,专门设置用户上下文到ThreadLocal中。伪代码如下:
  1. public class UserInterceptor extends HandlerInterceptorAdapter {
  2.    @Override
  3.    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
  4.       CurrentUser user = getUser(request);
  5.       if(Objects.nonNull(user)) {
  6.          CurrentUser.set(user);
  7.       }
  8.    }
  9. }
复制代码
用户在请求我们接口时,会先触发该拦截器,它会根据用户cookie中的token,调用调用接口获取redis中的用户信息。假如能获取到,阐明用户已经登录,则把用户信息设置到CurrentUser类的ThreadLocal中。
接下来,在api服务的下层,即business层的方法中,就能轻松通过CurrentUser.getCurrent();方法获取到想要的用户上下文信息了。

这套用户体系的想法是很good的,但深入使用后,发现了一个小插曲:
api服务和mq消费者服务都引用了business层,business层中的方法两个服务都能直接调用。
我们都知道在api服务中用户是需要登录的,而mq消费者服务则不需要登录。

假如business中的某个方法刚开始是给api开发的,在方法深处使用了CurrentUser.getCurrent();获取用户上下文。但后来,某位新来的帅哥在mq消费者中也调用了那个方法,并未发觉这个小构造,就会中招,出现找不到用户上下文的标题。

以是我其时的第一个想法是:代码没做兼容处理处罚,因为之前这类标题偶尔会发生一次。
想要解决这个标题,其实也很简朴。只需先判断一下可否从CurrentUser中获取用户信息,假如不能,则取设置的体系用户信息。伪代码如下:
  1. @Autowired
  2. private BusinessConfig businessConfig;
  3. CurrentUser user = CurrentUser.getCurrent();
  4. if(Objects.nonNull(user)) {
  5.    entity.setUserId(user.getUserId());
  6.    entity.setUserName(user.getUserName());
  7. } else {
  8.    entity.setUserId(businessConfig.getDefaultUserId());
  9.    entity.setUserName(businessConfig.getDefaultUserName());
  10. }
复制代码
这种简朴无公害的代码,假如只是在一两个地方加还OK。
此外,众所周知,SimpleDateFormat在java8以前,是用来处理处罚时间的工具类,它是非线程安全的。也就是说,用该方法分析日期会有线程安全标题。
为了避免线程安全标题的出现,我们可以把SimpleDateFormat对象界说成局部变量。但假如你肯定要把它界说成静态变量,可以使用ThreadLocal保存日期,也能解决线程安全标题。
8. 传递参数

之前见过有些同事写代码时,一个非常风趣的用法,即:使用MDC传递参数。
MDC是什么?
MDC是org.slf4j包下的一个类,它的全称是Mapped Diagnostic Context,我们可以认为它是一个线程安全的存放诊断日志的容器。
MDC的底层是用了ThreadLocal来保存数据的。
例如现在有这样一种场景:我们使用RestTemplate调用远程接口时,有时需要在header中传递信息,好比:traceId,source等,便于在查询日志时能够串联一次完备的请求链路,快速定位标题。
这种业务场景就能通过ClientHttpRequestInterceptor接口实现,详细做法如下:
第一步,界说一个LogFilter拦截所有接口请求,在MDC中设置traceId:
  1. public class LogFilter implements Filter {
  2.     @Override
  3.     public void init(FilterConfig filterConfig) throws ServletException {
  4.     }
  5.     @Override
  6.     public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
  7.         MdcUtil.add(UUID.randomUUID().toString());
  8.         System.out.println("记录请求日志");
  9.         chain.doFilter(request, response);
  10.         System.out.println("记录响应日志");
  11.     }
  12.     @Override
  13.     public void destroy() {
  14.     }
  15. }
复制代码
第二步,实现ClientHttpRequestInterceptor接口,MDC中获取当前请求的traceId,然后设置到header中:
  1. public class RestTemplateInterceptor implements ClientHttpRequestInterceptor {
  2.     @Override
  3.     public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {
  4.         request.getHeaders().set("traceId", MdcUtil.get());
  5.         return execution.execute(request, body);
  6.     }
  7. }
复制代码
第三步,界说设置类,设置上面界说的RestTemplateInterceptor类:
  1. @Configuration
  2. public class RestTemplateConfiguration {
  3.     @Bean
  4.     public RestTemplate restTemplate() {
  5.         RestTemplate restTemplate = new RestTemplate();
  6.         restTemplate.setInterceptors(Collections.singletonList(restTemplateInterceptor()));
  7.         return restTemplate;
  8.     }
  9.     @Bean
  10.     public RestTemplateInterceptor restTemplateInterceptor() {
  11.         return new RestTemplateInterceptor();
  12.     }
  13. }
复制代码
其中MdcUtil其实是使用MDC工具在ThreadLocal中存储和获取traceId
  1. public class MdcUtil {
  2.     private static final String TRACE_ID = "TRACE_ID";
  3.     public static String get() {
  4.         return MDC.get(TRACE_ID);
  5.     }
  6.     public static void add(String value) {
  7.         MDC.put(TRACE_ID, value);
  8.     }
  9. }
复制代码
固然,这个例子中没有演示MdcUtil类的add方法详细调的地方,我们可以在filter中实行接口方法之前,天生traceId,调用MdcUtil类的add方法添加到MDC中,然后在同一个请求的其他地方就能通过MdcUtil类的get方法获取到该traceId。
能使用MDC保存traceId等参数的根本原因是,用户请求到应用服务器,Tomcat会从线程池中分配一个线程去处理处罚该请求。
那么该请求的整个过程中,保存到MDC的ThreadLocal中的参数,也是该线程独享的,以是不会有线程安全标题。
9. 模拟高并发

有时候我们写的接口,在低并发的场景下,一点标题都没有。
但假如一旦出现高并发调用,该接口可能会出现一些意想不到的标题。
为了防止雷同的事情发生,一样平常在项目上线前,我们非常有须要对接口做一下压力测试。
固然,现在已经有比力成熟的压力测试工具,好比:Jmeter、LoadRunner等。
假如你觉得下载压测工具比力麻烦,也可以手写一个简朴的模拟并发操作的工具,用CountDownLatch就能实现,例如:
  1. public static void concurrenceTest() {
  2. /**
  3.      * 模拟高并发情况代码
  4.      */
  5.     final AtomicInteger atomicInteger = new AtomicInteger(0);
  6.     final CountDownLatch countDownLatch = new CountDownLatch(1000);// 相当于计数器,当所有都准备好了,再一起执行,模仿多并发,保证并发量
  7.     final CountDownLatch countDownLatch2 = new CountDownLatch(1000);// 保证所有线程执行完了再打印atomicInteger的值
  8.     ExecutorService executorService = Executors.newFixedThreadPool(10);
  9.     try {
  10.         for (int i = 0; i < 1000; i++) {
  11.             executorService.submit(new Runnable() {
  12.                 @Override
  13.                 public void run() {
  14.                     try {
  15.                         countDownLatch.await();//一直阻塞当前线程,直到计时器的值为0,保证同时并发
  16.                     } catch (InterruptedException e) {
  17.                         log.error(e.getMessage(),e);
  18.                     }
  19. //每个线程增加1000次,每次加1
  20.                     for (int j = 0; j < 1000; j++) {
  21.                         atomicInteger.incrementAndGet();
  22.                     }
  23.                     countDownLatch2.countDown();
  24.                 }
  25.             });
  26.             countDownLatch.countDown();
  27.         }
  28.         countDownLatch2.await();// 保证所有线程执行完
  29.         executorService.shutdown();
  30.     } catch (Exception e){
  31.         log.error(e.getMessage(),e);
  32.     }
  33. }
复制代码
10. 处理处罚mq消息

在高并发的场景中,消息积存标题,可以说跬步不离,真的没办法从根本上解决。外貌上看,已经解决了,但后面不知道什么时候,就会冒出一次,好比这次:
有天下战书,产品过来说:有几个商户投诉过来了,他们说菜品有延迟,快查一下原因。
这次标题出现得有点奇怪。
为什么这么说?
首先这个时间点就有点奇怪,平常出标题,不都是中午或者晚上用餐高峰期吗?怎么这次标题出现在下战书?
根据以往积聚的经验,我直接看了kafka的topic的数据,果然上面消息有积存,但这次每个partition都积存了十几万的消息没有消费,比以往加压的消息数目增加了几百倍。这次消息积存得极不寻常。
我赶紧查服务监控看看消费者挂了没,还好没挂。又查服务日志没有发现异常。这时我有点迷茫,碰运气问了问订单组下战书发生了什么事情没?他们说下战书有个促销活动,跑了一个JOB批量更新过有些商户的订单信息。
这时,我一下子如梦初醒,是他们在JOB中批量发消息导致的标题。怎么没有通知我们呢?实在太坑了。
虽说知道标题的原因了,倒是眼前积存的这十几万的消息该如何处理处罚呢?
此时,假如直接调大partition数目是不行的,汗青消息已经存储到4个固定的partition,只有新增的消息才会到新的partition。我们重点需要处理处罚的是已有的partition。
直接加服务节点也不行,因为kafka答应同组的多个partition被一个consumer消费,但不答应一个partition被同组的多个consumer消费,可能会造成资源浪费。
看来只有效多线程处理处罚了。
为了紧急解决标题,我改成了用线程池处理处罚消息,焦点线程和最大线程数都设置成了50。
大抵用法如下:

  • 先界说一个线程池:
  1. @Configuration
  2. public class ThreadPoolConfig {
  3.     @Value("${thread.pool.corePoolSize:5}")
  4.     private int corePoolSize;
  5.     @Value("${thread.pool.maxPoolSize:10}")
  6.     private int maxPoolSize;
  7.     @Value("${thread.pool.queueCapacity:200}")
  8.     private int queueCapacity;
  9.     @Value("${thread.pool.keepAliveSeconds:30}")
  10.     private int keepAliveSeconds;
  11.     @Value("${thread.pool.threadNamePrefix:ASYNC_}")
  12.     private String threadNamePrefix;
  13.     @Bean("messageExecutor")
  14.     public Executor messageExecutor() {
  15.         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  16.         executor.setCorePoolSize(corePoolSize);
  17.         executor.setMaxPoolSize(maxPoolSize);
  18.         executor.setQueueCapacity(queueCapacity);
  19.         executor.setKeepAliveSeconds(keepAliveSeconds);
  20.         executor.setThreadNamePrefix(threadNamePrefix);
  21.         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  22.         executor.initialize();
  23.         return executor;
  24.     }
  25. }
复制代码

  • 再界说一个消息的consumer:
  1. @Service
  2. public class MyConsumerService {
  3.     @Autowired
  4.     private Executor messageExecutor;
  5.     @KafkaListener(id="test",topics={"topic-test"})
  6.     public void listen(String message){
  7.         System.out.println("收到消息:" + message);
  8.         messageExecutor.submit(new MyWork(message);
  9.     }
  10. }
复制代码

  • 在界说的Runable实现类中处理处罚业务逻辑:
  1. public class MyWork implements Runnable {
  2.     private String message;
  3.     public MyWork(String message) {
  4.        this.message = message;
  5.     }
  6.     @Override
  7.     public void run() {
  8.         System.out.println(message);
  9.     }
  10. }
复制代码
果然,调解之后消息积存数目确实下降的非常快,大约半小时后,积存的消息就非常顺遂的处理处罚完了。
但此时有个更严峻的标题出现:我收到了报警邮件,有两个订单体系的节点down机了。。。
11. 统计数目

在多线程的场景中,有时候需要统计数目,好比:用多线程导入供应商数据时,统计导入成功的供应商数有多少。
假如这时候用count++统计次数,终极的结果可能会禁绝。因为count++并非原子操作,假如多个线程同时实行该操作,则统计的次数,可能会出现异常。
为了解决这个标题,就需要使用concurent的atomic包下面的类,好比:AtomicInteger、AtomicLong等。
  1. @Servcie
  2. public class ImportSupplierService {
  3.   private static AtomicInteger count = new AtomicInteger(0);
  4.   public int importSupplier(List<SupplierInfo> supplierList) {
  5.        if(CollectionUtils.isEmpty(supplierList)) {
  6.            return 0;
  7.        }
  8.        supplierList.parallelStream().forEach(x -> {
  9.            try {
  10.              importSupplier(x);
  11.              count.addAndGet(1);
  12.            } catch(Exception e) {
  13.               log.error(e.getMessage(),e);
  14.            }
  15.        );
  16.       return count.get();
  17.   }
  18. }
复制代码
AtomicInteger的底层说白了使用自旋锁+CAS。
  1. public final int incrementAndGet() {
  2.     for (;;) {
  3.         int current = get();
  4.         int next = current + 1;
  5.         if (compareAndSet(current, next))
  6.             return next;
  7.     }
  8. }
复制代码
自旋锁说白了就是一个死循环。
而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,表示以固定延时实行任务,延时是相对当前任务结束为出发点计算开始时间。
实现这种定时任务的详细代码如下:
  1. public class ScheduleExecutorTest {
  2.     public static void main(String[] args) {
  3.         ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
  4.         scheduledExecutorService.scheduleAtFixedRate(() -> {
  5.             System.out.println("doSomething");
  6.         },1000,1000, TimeUnit.MILLISECONDS);
  7.     }
  8. }
复制代码
调用ScheduledExecutorService类的scheduleAtFixedRate方法实现周期性任务,每隔1秒钟实行一次,每次延迟1秒再实行。
这种定时任务是阿里巴巴开发者规范中用来替换Timer类的方案,对于多线程实行周期性任务,是个不错的选择。
使用ScheduledExecutorService类做延迟定时任务的优缺点:


  • 优点:基于多线程的定时任务,多个任务之间不会相关影响,支持周期性的实行任务,而且带延迟功能。
  • 缺点:不支持一些较复杂的定时规则。
固然,你也可以使用分布式定时任务,好比:xxl-job或者elastic-job等等。
其实,在现实工作中我使用多线程的场景远远不只这12种,在这里只是抛砖引玉,介绍了一些我认为比力常见的业务场景。

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有账号?立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

万万哇

金牌会员
这个人很懒什么都没写!

标签云

快速回复 返回顶部 返回列表