我的博客

想必大家都晓得，并发编程是一项十分重要的技术，无论在面试，还是工作中呈现的频率十分高。

当然，如果是集体级别的小玩意，必定不是特地在意这个。

并发编程 == 多线程编程。

然而多线程肯定比单线程效率更高？

不肯定，得看具体的业务环境，多个线程之间，会一直的抢占CPU的资源，所以说有时候多线程也是绝对耗费工夫的。

并发编程场景

1.定时工作

你可能纳闷？Thread类真的能做定时工作吗？

答案是必定的。一些定时工作框架的源码，它们的底层也会应用Thread类。
 public static void init() {     new Thread(() -> {         while (true) {             try {                 System.out.println("下载文件");                 Thread.sleep(1000 * 60 * 5);             } catch (Exception e) {                 log.error(e);             }         }     }).start(); }

应用Thread类能够做最简略的定时工作，在run办法中有个while的死循环（当然还有其余形式），执行咱们本人的工作。有个须要特地留神的中央是，须要用try...catch捕捉异样，否则如果出现异常，就间接退出循环，下次将无奈继续执行了。

但这种形式做的定时工作，只能周期性执行，不能反对定时在某个工夫点执行。

特地揭示一下，该线程倡议定义成守护线程，能够通过setDaemon办法设置，让它在后盾默默执行就好。

应用场景：比方我的项目中有时须要每隔5分钟去下载某个文件，或者每隔10分钟去读取模板文件生成动态html页面等等，一些简略的周期性工作场景。

应用Thread类做定时工作的优缺点：

  • 长处：这种定时工作非常简单，学习成本低，容易动手，对于那些简略的周期性工作，是个不错的抉择。
  • 毛病：不反对指定某个工夫点执行工作，不反对提早执行等操作，性能过于繁多，无奈应答一些较为简单的场景。

2.监听器

有时候，咱们须要写个监听器，去监听某些数据的变动。

比方：咱们在应用canal的时候，须要监听binlog的变动，可能及时把数据库中的数据，同步到另外一个业务数据库中。

如果间接写一个监听器去监听数据就太没意思了，咱们想实现这样一个性能：在配置核心有个开关，配置监听器是否开启，如果开启了应用单线程异步执行。

次要代码如下：

 @Service public CanalService {     private volatile boolean running = false;     private Thread thread;      @Autowired     private CanalConnector canalConnector;          public void handle() {         //连贯canal         while(running) {            //业务解决         }     }          public void start() {        thread = new Thread(this::handle, "name");        running = true;        thread.start();     }          public void stop() {        if(!running) {           return;        }        running = false;     } }

在start办法中开启了一个线程，在该线程中异步执行handle办法的具体任务。而后通过调用stop办法，能够进行该线程。

其中，应用volatile关键字管制的running变量作为开关，它能够控制线程中的状态。

接下来，有个比拟要害的点是：如何通过配置核心的配置，管制这个开关呢？

以apollo配置为例，咱们在配置核心的后盾，批改配置之后，主动获取最新配置的外围代码如下：

 public class CanalConfig {     @Autowired     private CanalService canalService;      @ApolloConfigChangeListener     public void change(ConfigChangeEvent event) {         String value = event.getChange("test.canal.enable").getNewValue();         if(BooleanUtils.toBoolean(value)) {             canalService.start();         } else {             canalService.stop();         }     } }

通过apollo的ApolloConfigChangeListener注解，能够监听配置参数的变动。

如果test.canal.enable开关配置的true，则调用canalService类的start办法开启canal数据同步性能。如果开关配置的false，则调用canalService类的stop办法，主动进行canal数据同步性能。

3.收集日志

在某些高并发的场景中，咱们须要收集局部用户的日志（比方：用户登录的日志），写到数据库中，以便于做剖析。

但因为我的项目中，还没有引入消息中间件，比方：kafka、rocketmq等。

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

所以，大家很天然想到了异步解决。

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

这样做，能够是能够。

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

其实，还有更优雅的解决形式，虽说没有应用消息中间件，但借用了它的思维。

这套记录登录日志的性能，分为：日志生产端、日志存储端和日志生产端。

先定义了一个阻塞队列。

 @Component public class LoginLogQueue {     private static final int QUEUE_MAX_SIZE    = 1000;      private BlockingQueueblockingQueue queue = new LinkedBlockingQueue<>(QUEUE_MAX_SIZE);      //生成音讯     public boolean push(LoginLog loginLog) {         return this.queue.add(loginLog);     }       //生产音讯     public LoginLog poll() {         LoginLog loginLog = null;         try {             loginLog = this.queue.take();         } catch (InterruptedException e) {             e.printStackTrace();         }         return result;     } }

而后定义了一个日志的生产者。

 @Service public class LoginSerivce {          @Autowired     private LoginLogQueue loginLogQueue;      public int login(UserInfo userInfo) {         //业务解决         LoginLog loginLog = convert(userInfo);         loginLogQueue.push(loginLog);     }   }

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

 @Service public class LoginInfoConsumer {     @Autowired     private LoginLogQueue queue;      @PostConstruct     public voit init {        new Thread(() -> {           while (true) {               LoginLog loginLog = queue.take();               //写入数据库           }         }).start();     } }

当然，这个例子中应用单线程接管登录日志，为了晋升性能，也能够应用线程池来解决业务逻辑（比方：写入数据库）等。

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的原理：

当然除了excel导入之外，还有相似的读取文本文件，也能够用相似的办法解决。

舒适的揭示一下，如果一次性导入的数据十分多，用多线程解决，可能会使零碎的CPU使用率飙升，须要特地关注。

5.查问接口

很多时候，咱们须要在某个查问接口中，调用其余服务的接口，组合数据之后，一起返回。

比方有这样的业务场景：

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

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

于是，用户信息查问接口须要调用用户查问接口、积分查问接口 和 成长值查问接口，而后汇总数据对立返回。

调用过程如下图所示：

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

显然这种串行调用近程接口性能是十分不好的，调用近程接口总的耗时为所有的近程接口耗时之和。

那么如何优化近程接口性能呢？

既然串行调用多个近程接口性能很差，为什么不改成并行呢？

如下图所示：

调用近程接口总耗时 200ms = 200ms（即耗时最长的那次近程接口调用）

在java8之前能够通过实现Callable接口，获取线程返回后果。

java8当前通过CompleteFuture类实现该性能。咱们这里以CompleteFuture为例：

 public UserInfo getUserInfo(Long id) throws InterruptedException, ExecutionException {     final UserInfo userInfo = new UserInfo();     CompletableFuture userFuture = CompletableFuture.supplyAsync(() -> {         getRemoteUserAndFill(id, userInfo);         return Boolean.TRUE;     }, executor);      CompletableFuture bonusFuture = CompletableFuture.supplyAsync(() -> {         getRemoteBonusAndFill(id, userInfo);         return Boolean.TRUE;     }, executor);      CompletableFuture growthFuture = CompletableFuture.supplyAsync(() -> {         getRemoteGrowthAndFill(id, userInfo);         return Boolean.TRUE;     }, executor);     CompletableFuture.allOf(userFuture, bonusFuture, growthFuture).join();      userFuture.get();     bonusFuture.get();     growthFuture.get();     return userInfo; }

舒适揭示一下，这两种形式别忘了应用线程池。示例中我用到了executor，示意自定义的线程池，为了避免高并发场景下，呈现线程过多的问题。

6.获取用户上下文

不晓得你在我的项目开发时，有没有遇到过这样的需要：用户登录之后，在所有的申请接口中，通过某个公共办法，就能获取到以后登录用户的信息？

获取的用户上下文，咱们以CurrentUser为例。

CurrentUser外部蕴含了一个ThreadLocal对象，它负责保留以后线程的用户上下文信息。当然为了保障在线程池中，也能从用户上下文中获取到正确的用户信息，这里用了阿里的TransmittableThreadLocal。伪代码如下：

 @Data public class CurrentUser {     private static final TransmittableThreadLocal<CurrentUser> THREA_LOCAL = new TransmittableThreadLocal<>();          private String id;     private String userName;     private String password;     private String phone;     ...          public statis void set(CurrentUser user) {       THREA_LOCAL.set(user);     }          public static void getCurrent() {       return THREA_LOCAL.get();     } }

这里为什么用了阿里的TransmittableThreadLocal，而不是一般的ThreadLocal呢？在线程池中，因为线程会被屡次复用，导致从一般的ThreadLocal中无奈获取正确的用户信息。父线程中的参数，没法传递给子线程，而TransmittableThreadLocal很好解决了这个问题。

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

 public class UserInterceptor extends HandlerInterceptorAdapter {        @Override      public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {       CurrentUser user = getUser(request);       if(Objects.nonNull(user)) {          CurrentUser.set(user);       }    }  }

用户在申请咱们接口时，会先触发该拦截器，它会依据用户cookie中的token，调用调用接口获取redis中的用户信息。如果能获取到，阐明用户曾经登录，则把用户信息设置到CurrentUser类的ThreadLocal中。

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

这套用户体系的想法是很good的，但深刻应用后，发现了一个小插曲：

api服务和mq消费者服务都援用了business层，business层中的办法两个服务都能间接调用。

咱们都晓得在api服务中用户是须要登录的，而mq消费者服务则不须要登录。

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

所以我过后的第一个想法是：代码没做兼容解决，因为之前这类问题偶然会产生一次。

想要解决这个问题，其实也很简略。只需先判断一下是否从CurrentUser中获取用户信息，如果不能，则取配置的零碎用户信息。伪代码如下：

 @Autowired private BusinessConfig businessConfig;  CurrentUser user = CurrentUser.getCurrent(); if(Objects.nonNull(user)) {    entity.setUserId(user.getUserId());    entity.setUserName(user.getUserName()); } else {    entity.setUserId(businessConfig.getDefaultUserId());    entity.setUserName(businessConfig.getDefaultUserName()); }

这种简略无公害的代码，如果只是在一两个中央加还OK。

此外，家喻户晓，SimpleDateFormat在java8以前，是用来解决工夫的工具类，它是非线程平安的。也就是说，用该办法解析日期会有线程平安问题。

为了防止线程平安问题的呈现，咱们能够把SimpleDateFormat对象定义成局部变量。但如果你肯定要把它定义成动态变量，能够应用ThreadLocal保留日期，也能解决线程平安问题。

8. 传递参数

之前见过有些共事写代码时，一个十分乏味的用法，即：应用MDC传递参数。

MDC是什么？

MDC是org.slf4j包下的一个类，它的全称是Mapped Diagnostic Context，咱们能够认为它是一个线程平安的寄存诊断日志的容器。

MDC的底层是用了ThreadLocal来保留数据的。

例如当初有这样一种场景：咱们应用RestTemplate调用近程接口时，有时须要在header中传递信息，比方：traceId，source等，便于在查问日志时可能串联一次残缺的申请链路，疾速定位问题。

这种业务场景就能通过ClientHttpRequestInterceptor接口实现，具体做法如下：

第一步，定义一个LogFilter拦挡所有接口申请，在MDC中设置traceId：

 public class LogFilter implements Filter {     @Override     public void init(FilterConfig filterConfig) throws ServletException {     }      @Override     public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {         MdcUtil.add(UUID.randomUUID().toString());         System.out.println("记录申请日志");         chain.doFilter(request, response);         System.out.println("记录响应日志");     }      @Override     public void destroy() {     } }

第二步，实现ClientHttpRequestInterceptor接口，MDC中获取以后申请的traceId，而后设置到header中：

 public class RestTemplateInterceptor implements ClientHttpRequestInterceptor {      @Override     public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) throws IOException {         request.getHeaders().set("traceId", MdcUtil.get());         return execution.execute(request, body);     } }

第三步，定义配置类，配置下面定义的RestTemplateInterceptor类：

 @Configuration public class RestTemplateConfiguration {      @Bean     public RestTemplate restTemplate() {         RestTemplate restTemplate = new RestTemplate();         restTemplate.setInterceptors(Collections.singletonList(restTemplateInterceptor()));         return restTemplate;     }      @Bean     public RestTemplateInterceptor restTemplateInterceptor() {         return new RestTemplateInterceptor();     } }

其中MdcUtil其实是利用MDC工具在ThreadLocal中存储和获取traceId

 public class MdcUtil {      private static final String TRACE_ID = "TRACE_ID";      public static String get() {         return MDC.get(TRACE_ID);     }      public static void add(String value) {         MDC.put(TRACE_ID, value);     } }

当然，这个例子中没有演示MdcUtil类的add办法具体调的中央，咱们能够在filter中执行接口办法之前，生成traceId，调用MdcUtil类的add办法增加到MDC中，而后在同一个申请的其余中央就能通过MdcUtil类的get办法获取到该traceId。

能应用MDC保留traceId等参数的根本原因是，用户申请到应用服务器，Tomcat会从线程池中调配一个线程去解决该申请。

那么该申请的整个过程中，保留到MDC的ThreadLocal中的参数，也是该线程独享的，所以不会有线程平安问题。

9. 模仿高并发

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

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

为了避免相似的事件产生，个别在我的项目上线前，咱们十分有必要对接口做一下压力测试。

当然，当初曾经有比拟成熟的压力测试工具，比方：Jmeter、LoadRunner等。

如果你感觉下载压测工具比拟麻烦，也能够手写一个简略的模仿并发操作的工具，用CountDownLatch就能实现，例如：

 public static void concurrenceTest() {     /**      * 模仿高并发状况代码      */     final AtomicInteger atomicInteger = new AtomicInteger(0);     final CountDownLatch countDownLatch = new CountDownLatch(1000); // 相当于计数器，当所有都筹备好了，再一起执行，模拟多并发，保障并发量     final CountDownLatch countDownLatch2 = new CountDownLatch(1000); // 保障所有线程执行完了再打印atomicInteger的值     ExecutorService executorService = Executors.newFixedThreadPool(10);     try {         for (int i = 0; i < 1000; i++) {             executorService.submit(new Runnable() {                 @Override                 public void run() {                     try {                         countDownLatch.await(); //始终阻塞以后线程，直到计时器的值为0,保障同时并发                     } catch (InterruptedException e) {                         log.error(e.getMessage(),e);                     }                     //每个线程减少1000次，每次加1                     for (int j = 0; j < 1000; j++) {                         atomicInteger.incrementAndGet();                     }                     countDownLatch2.countDown();                 }             });             countDownLatch.countDown();         }          countDownLatch2.await();// 保障所有线程执行完         executorService.shutdown();     } catch (Exception e){         log.error(e.getMessage(),e);     } }

10. 解决mq音讯

在高并发的场景中，音讯积压问题，能够说如影随形，真的没方法从根本上解决。外表上看，曾经解决了，但前面不晓得什么时候，就会冒出一次，比方这次：

有天下午，产品过去说：有几个商户投诉过去了，他们说菜品有提早，快查一下起因。

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

为什么这么说？

首先这个工夫点就有点奇怪，平时出问题，不都是中午或者早晨用餐高峰期吗？怎么这次问题呈现在下午？

依据以往积攒的教训，我间接看了kafka的topic的数据，果然下面音讯有积压，但这次每个partition都积压了十几万的音讯没有生产，比以往加压的音讯数量减少了几百倍。这次音讯积压得极不寻常。

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

这时，我一下子如梦初醒，是他们在JOB中批量发消息导致的问题。怎么没有告诉咱们呢？切实太坑了。

虽说晓得问题的起因了，倒是眼前积压的这十几万的音讯该如何解决呢？

此时，如果间接调大partition数量是不行的，历史音讯曾经存储到4个固定的partition，只有新增的音讯才会到新的partition。咱们重点须要解决的是已有的partition。

间接加服务节点也不行，因为kafka容许同组的多个partition被一个consumer生产，但不容许一个partition被同组的多个consumer生产，可能会造成资源节约。

看来只有用多线程解决了。

为了紧急解决问题，我改成了用线程池解决音讯，外围线程和最大线程数都配置成了50。

大抵用法如下：

  1. 先定义一个线程池：
 @Configuration public class ThreadPoolConfig {      @Value("${thread.pool.corePoolSize:5}")     private int corePoolSize;      @Value("${thread.pool.maxPoolSize:10}")     private int maxPoolSize;      @Value("${thread.pool.queueCapacity:200}")     private int queueCapacity;      @Value("${thread.pool.keepAliveSeconds:30}")     private int keepAliveSeconds;      @Value("${thread.pool.threadNamePrefix:ASYNC_}")     private String threadNamePrefix;      @Bean("messageExecutor")     public Executor messageExecutor() {         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();         executor.setCorePoolSize(corePoolSize);         executor.setMaxPoolSize(maxPoolSize);         executor.setQueueCapacity(queueCapacity);         executor.setKeepAliveSeconds(keepAliveSeconds);         executor.setThreadNamePrefix(threadNamePrefix);         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());         executor.initialize();         return executor;     } }
  1. 再定义一个音讯的consumer：
 @Service public class MyConsumerService {     @Autowired     private Executor messageExecutor;          @KafkaListener(id="test",topics={"topic-test"})     public void listen(String message){         System.out.println("收到音讯：" + message);         messageExecutor.submit(new MyWork(message);     } }
  1. 在定义的Runable实现类中解决业务逻辑：
 public class MyWork implements Runnable {     private String message;          public MyWork(String message) {        this.message = message;     }      @Override     public void run() {         System.out.println(message);     } }

果然，调整之后音讯积压数量的确降落的十分快，大概半小时后，积压的音讯就十分顺利的解决完了。

但此时有个更重大的问题呈现：我收到了报警邮件，有两个订单零碎的节点宕机了。。。

11. 统计数量

在多线程的场景中，有时候须要统计数量，比方：用多线程导入供应商数据时，统计导入胜利的供应商数有多少。

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

为了解决这个问题，就须要应用concurent的atomic包上面的类，比方：AtomicInteger、AtomicLong等。

 @Servcie public class ImportSupplierService {   private static AtomicInteger count = new AtomicInteger(0);    public int importSupplier(List<SupplierInfo> supplierList) {        if(CollectionUtils.isEmpty(supplierList)) {            return 0;        }         supplierList.parallelStream().forEach(x -> {            try {              importSupplier(x);              count.addAndGet(1);            } catch(Exception e) {               log.error(e.getMessage(),e);            }        );        return count.get();   }     }

AtomicInteger的底层说白了应用自旋锁+CAS。

 public final int incrementAndGet() {     for (;;) {         int current = get();         int next = current + 1;         if (compareAndSet(current, next))             return next;     } }

自旋锁说白了就是一个死循环。

而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，示意以固定延时执行工作，延时是绝对当前任务完结为终点计算开始工夫。

实现这种定时工作的具体代码如下：

 public class ScheduleExecutorTest {      public static void main(String[] args) {         ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);         scheduledExecutorService.scheduleAtFixedRate(() -> {             System.out.println("doSomething");         },1000,1000, TimeUnit.MILLISECONDS);     } }

调用ScheduledExecutorService类的scheduleAtFixedRate办法实现周期性工作，每隔1秒钟执行一次，每次提早1秒再执行。

这种定时工作是阿里巴巴开发者标准中用来代替Timer类的计划，对于多线程执行周期性工作，是个不错的抉择。

应用ScheduledExecutorService类做提早定时工作的优缺点：

  • 长处：基于多线程的定时工作，多个工作之间不会相干影响，反对周期性的执行工作，并且带提早性能。
  • 毛病：不反对一些较简单的定时规定。

当然，你也能够应用分布式定时工作，比方：xxl-job或者elastic-job等等。

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

文章来自：苏三说技术