• 并发编程的12种业务场景


    前言

    并发编程是一项非常重要的技术,无论在面试,还是工作中出现的频率非常高。

    继续聊聊并发编程这个话题。

    并发编程说白了就是多线程编程,但多线程一定比单线程效率更高?

    答:不一定,要看具体业务场景。

    毕竟如果使用了多线程,那么线程之间的竞争和抢占cpu资源,线程的上下文切换,也是相对来说比较耗时的操作。

    下面这几个问题在面试中,你必定遇到过:

    1. 你在哪来业务场景中使用过多线程?

    2. 怎么用的?

    3. 踩过哪些坑?

    今天聊聊我之前在项目中用并发编程的12种业务场景,给有需要的朋友一个参考。

    c2b3d5f0becf4e19bc351d0957d7b7c5.png 

    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的变化,能够及时把数据库中的数据,同步到另外一个业务数据库中。

    ba47d8c670b14710866136077a85229d.png 

    如果直接写一个监听器去监听数据就太没意思了,我们想实现这样一个功能:在配置中心有个开关,配置监听器是否开启,如果开启了使用单线程异步执行。

    主要代码如下:

    1. @Service
    2. public CanalService {
    3.     private volatile boolean running = false;
    4.     private Thread thread;
    5.     @Autowired
    6.     private CanalConnector canalConnector;
    7.     
    8.     public void handle() {
    9.         //连接canal
    10.         while(running) {
    11.            //业务处理
    12.         }
    13.     }
    14.     
    15.     public void start() {
    16.        thread = new Thread(this::handle, "name");
    17.        running = true;
    18.        thread.start();
    19.     }
    20.     
    21.     public void stop() {
    22.        if(!running) {
    23.           return;
    24.        }
    25.        running = false;
    26.     }
    27. }

    在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. }

    通过apolloApolloConfigChangeListener注解,可以监听配置参数的变化。

    如果test.canal.enable开关配置的true,则调用canalService类的start方法开启canal数据同步功能。如果开关配置的false,则调用canalService类的stop方法,自动停止canal数据同步功能。

    3.收集日志

    在某些高并发的场景中,我们需要收集部分用户的日志(比如:用户登录的日志),写到数据库中,以便于做分析。

    但由于项目中,还没有引入消息中间件,比如:kafkarocketmq等。

    如果直接将日志同步写入数据库,可能会影响接口性能。

    所以,大家很自然想到了异步处理。

    实现这个需求最简单的做法是,开启一个线程,异步写入数据到数据库即可。

    这样做,可以是可以。

    但如果用户登录操作的耗时,比异步写入数据库的时间要少得多。这样导致的结果是:生产日志的速度,比消费日志的速度要快得多,最终的性能瓶颈在消费端。

    其实,还有更优雅的处理方式,虽说没有使用消息中间件,但借用了它的思想。

    这套记录登录日志的功能,分为:日志生产端、日志存储端和日志消费端。

    如下图所示:

    8fd9e05ddb724d3fae344abf7629e0a3.png 

    先定义了一个阻塞队列。

    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.     
    4.     @Autowired
    5.     private LoginLogQueue loginLogQueue;
    6.     public int login(UserInfo userInfo) {
    7.         //业务处理
    8.         LoginLog loginLog = convert(userInfo);
    9.         loginLogQueue.push(loginLog);
    10.     }  
    11. }

    接下来,定义了日志的消费者。

    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数据导入需求,比如:将某一大类下的所有子类一次性导入系统,或者导入一批新的供应商数据等等。

    我们以导入供应商数据为例,它所涉及的业务流程很长,比如:

    1. 调用天眼查接口校验企业名称和统一社会信用代码。

    2. 写入供应商基本表

    3. 写入组织表

    4. 给供应商自动创建一个用户

    5. 给该用户分配权限

    6. 自定义域名

    7. 发站内通知

    等等。

    如果在程序中,解析完excel,读取了所有数据之后。用单线程一条条处理业务逻辑,可能耗时会非常长。

    为了提升excel数据导入效率,非常有必要使用多线程来处理。

    当然在java中实现多线程的手段有很多种,下面重点聊聊java8中最简单的实现方式:parallelStream

    伪代码如下:

    supplierList.parallelStream().forEach(x -> importSupplier(x));
    

    parallelStream是一个并行执行的流,它默认通过ForkJoinPool实现的,能提高你的多线程任务的速度。

    ForkJoinPool处理的过程会分而治之,它的核心思想是:将一个大任务切分成多个小任务。每个小任务都能单独执行,最后它会把所用任务的执行结果进行汇总。

    下面用一张图简单介绍一下ForkJoinPool的原理:

    56d30c94f9f349bf92caf56e68b19225.png ma 

    当然除了excel导入之外,还有类似的读取文本文件,也可以用类似的方法处理。

    温馨的提醒一下,如果一次性导入的数据非常多,用多线程处理,可能会使系统的cpu使用率飙升,需要特别关注。

    5.查询接口

    很多时候,我们需要在某个查询接口中,调用其他服务的接口,组合数据之后,一起返回。

    比如有这样的业务场景:

    在用户信息查询接口中需要返回:用户名称、性别、等级、头像、积分、成长值等信息。

    而用户名称、性别、等级、头像在用户服务中,积分在积分服务中,成长值在成长值服务中。为了汇总这些数据统一返回,需要另外提供一个对外接口服务。

    于是,用户信息查询接口需要调用用户查询接口、积分查询接口 和 成长值查询接口,然后汇总数据统一返回。

    调用过程如下图所示:

     

    7f7d96d05e5b4dc9afc4d5eae0edee97.png 

    调用远程接口总耗时 530ms = 200ms + 150ms + 180ms

    显然这种串行调用远程接口性能是非常不好的,调用远程接口总的耗时为所有的远程接口耗时之和。

    那么如何优化远程接口性能呢?

    既然串行调用多个远程接口性能很差,为什么不改成并行呢?

    如下图所示:

     

    312c538447b642b7b96f526fcfc44932.png 

    调用远程接口总耗时 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<CurrentUserTHREA_LOCAL = new TransmittableThreadLocal<>();
    4.     
    5.     private String id;
    6.     private String userName;
    7.     private String password;
    8.     private String phone;
    9.     ...
    10.     
    11.     public statis void set(CurrentUser user) {
    12.       THREA_LOCAL.set(user);
    13.     }
    14.     
    15.     public static void getCurrent() {
    16.       return THREA_LOCAL.get();
    17.     }
    18. }

    这里为什么用了阿里的TransmittableThreadLocal,而不是普通的ThreadLocal呢?在线程池中,由于线程会被多次复用,导致从普通的ThreadLocal中无法获取正确的用户信息。父线程中的参数,没法传递给子线程,而TransmittableThreadLocal很好解决了这个问题。

    然后在项目中定义一个全局的spring mvc拦截器,专门设置用户上下文到ThreadLocal中。伪代码如下:

    1. public class UserInterceptor extends HandlerInterceptorAdapter {
    2.    
    3.    @Override  
    4.    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
    5.       CurrentUser user = getUser(request);
    6.       if(Objects.nonNull(user)) {
    7.          CurrentUser.set(user);
    8.       }
    9.    } 
    10. }

    用户在请求我们接口时,会先触发该拦截器,它会根据用户cookie中的token,调用调用接口获取redis中的用户信息。如果能获取到,说明用户已经登录,则把用户信息设置到CurrentUser类的ThreadLocal中。

    接下来,在api服务的下层,即business层的方法中,就能轻松通过CurrentUser.getCurrent();方法获取到想要的用户上下文信息了。

     

    fc732299d20c452dbe1701aa40dd90e7.png 

    这套用户体系的想法是很good的,但深入使用后,发现了一个小插曲:

    api服务和mq消费者服务都引用了business层,business层中的方法两个服务都能直接调用。

    我们都知道在api服务中用户是需要登录的,而mq消费者服务则不需要登录。

     

    5526f3b02a494ae5bf06ae694e23a853.png 

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

    821f611430c3473aa044014205249410.png  

    所以我当时的第一个想法是:代码没做兼容处理,因为之前这类问题偶尔会发生一次。

    想要解决这个问题,其实也很简单。只需先判断一下能否从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是什么?

    MDCorg.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会从线程池中分配一个线程去处理该请求。

    那么该请求的整个过程中,保存到MDCThreadLocal中的参数,也是该线程独享的,所以不会有线程安全问题。

    9. 模拟高并发

    有时候我们写的接口,在低并发的场景下,一点问题都没有。

    但如果一旦出现高并发调用,该接口可能会出现一些意想不到的问题。

    为了防止类似的事情发生,一般在项目上线前,我们非常有必要对接口做一下压力测试

    当然,现在已经有比较成熟的压力测试工具,比如:JmeterLoadRunner等。

    如果你觉得下载压测工具比较麻烦,也可以手写一个简单的模拟并发操作的工具,用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消息

    在高并发的场景中,消息积压问题,可以说如影随形,真的没办法从根本上解决。表面上看,已经解决了,但后面不知道什么时候,就会冒出一次,比如这次:

    有天下午,产品过来说:有几个商户投诉过来了,他们说菜品有延迟,快查一下原因。

    这次问题出现得有点奇怪。

    为什么这么说?

    首先这个时间点就有点奇怪,平常出问题,不都是中午或者晚上用餐高峰期吗?怎么这次问题出现在下午?

    根据以往积累的经验,我直接看了kafkatopic的数据,果然上面消息有积压,但这次每个partition都积压了十几万的消息没有消费,比以往加压的消息数量增加了几百倍。这次消息积压得极不寻常。

    我赶紧查服务监控看看消费者挂了没,还好没挂。又查服务日志没有发现异常。这时我有点迷茫,碰运气问了问订单组下午发生了什么事情没?他们说下午有个促销活动,跑了一个JOB批量更新过有些商户的订单信息。

    这时,我一下子如梦初醒,是他们在JOB中批量发消息导致的问题。怎么没有通知我们呢?实在太坑了。

    虽说知道问题的原因了,倒是眼前积压的这十几万的消息该如何处理呢?

    此时,如果直接调大partition数量是不行的,历史消息已经存储到4个固定的partition,只有新增的消息才会到新的partition。我们重点需要处理的是已有的partition。

    直接加服务节点也不行,因为kafka允许同组的多个partition被一个consumer消费,但不允许一个partition被同组的多个consumer消费,可能会造成资源浪费。

    看来只有用多线程处理了。

    为了紧急解决问题,我改成了用线程池处理消息,核心线程和最大线程数都配置成了50

    大致用法如下:

    1. 先定义一个线程池:

    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. }
    1. 再定义一个消息的consumer:

    1. @Service
    2. public class MyConsumerService {
    3.     @Autowired
    4.     private Executor messageExecutor;
    5.     
    6.     @KafkaListener(id="test",topics={"topic-test"})
    7.     public void listen(String message){
    8.         System.out.println("收到消息:" + message);
    9.         messageExecutor.submit(new MyWork(message);
    10.     }
    11. }
    1. 在定义的Runable实现类中处理业务逻辑:

    1. public class MyWork implements Runnable {
    2.     private String message;
    3.     
    4.     public MyWork(String message) {
    5.        this.message = message;
    6.     }
    7.     @Override
    8.     public void run() {
    9.         System.out.println(message);
    10.     }
    11. }

    果然,调整之后消息积压数量确实下降的非常快,大约半小时后,积压的消息就非常顺利的处理完了。

    但此时有个更严重的问题出现:我收到了报警邮件,有两个订单系统的节点down机了。

    11. 统计数量

    在多线程的场景中,有时候需要统计数量,比如:用多线程导入供应商数据时,统计导入成功的供应商数有多少。

    如果这时候用count++统计次数,最终的结果可能会不准。因为count++并非原子操作,如果多个线程同时执行该操作,则统计的次数,可能会出现异常。

    为了解决这个问题,就需要使用concurentatomic包下面的类,比如:AtomicIntegerAtomicLong等。

    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分钟还未完成支付,则系统自动将该订单取消。

    这里需求就可以使用延迟定时任务实现。

    ScheduledExecutorServiceJDK1.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,1000TimeUnit.MILLISECONDS);
    7.     }
    8. }

    调用ScheduledExecutorService类的scheduleAtFixedRate方法实现周期性任务,每隔1秒钟执行一次,每次延迟1秒再执行。

    这种定时任务是阿里巴巴开发者规范中用来替代Timer类的方案,对于多线程执行周期性任务,是个不错的选择。

    使用ScheduledExecutorService类做延迟定时任务的优缺点:

    • 优点:基于多线程的定时任务,多个任务之间不会相关影响,支持周期性的执行任务,并且带延迟功能。

    • 缺点:不支持一些较复杂的定时规则。

    当然,你也可以使用分布式定时任务,比如:xxl-job或者elastic-job等等。

    其实,在实际工作中我使用多线程的场景远远不只这12种,在这里只是抛砖引玉,介绍了一些我认为比较常见的业务场景。 

     

  • 相关阅读:
    Web进阶
    如何快速提升教育直播间人气
    汇编语言王爽第四版检测点15.1答案
    机器学习(二)线性回归
    BUUCTF Misc 来首歌吧 & 荷兰宽带数据泄露 & 面具下的flag & 九连环
    Could not find com.android.tools.build:gradle:2.3.1
    程序员搞开源,读什么书最合适?
    List<Map<String, String>>数据行转列
    DataCamp在线学习平台
    【C++ 学习 ㊳】- 详解 C++ 强制类型转换
  • 原文地址:https://blog.csdn.net/m0_72088858/article/details/126764027