使用多线程(2020-03-09)

有两种多线程的使用方式

一. @Async的使用

1. 何为异步调用？

在解释异步调用之前，我们先来看同步调用的定义；同步就是整个处理过程顺序执行，当各个过程都执行完毕，并返回结果。 异步调用则是只是发送了调用的指令，调用者无需等待被调用的方法完全执行完毕；而是继续执行下面的流程。

例如， 在某个调用中，需要顺序调用 A, B, C三个过程方法；如他们都是同步调用，则需要将他们都顺序执行完毕之后，方算作过程执行完毕； 如B为一个异步的调用方法，则在执行完A之后，调用B，并不等待B完成，而是执行开始调用C，待C执行完毕之后，就意味着这个过程执行完毕了。
2. 常规的异步调用处理方式

在Java中，一般在处理类似的场景之时，都是基于创建独立的线程去完成相应的异步调用逻辑，通过主线程和不同的线程之间的执行流程，从而在启动独立的线程之后，主线程继续执行而不会产生停滞等待的情况。

3. @Async介绍

在Spring中，基于@Async标注的方法，称之为异步方法；这些方法将在执行的时候，将会在独立的线程中被执行，调用者无需等待它的完成，即可继续其他的操作。

4.如何在Spring中启用@Async
  • <font size = 2>基于Java配置的启用方式： 
 @Configuration
 @EnableAsync
 public class SpringAsyncConfig { ... }
  • <font size = 2>在启动类中的启用方式(spring cloud 微服务) 
@EnableAsync
public class SjServiceJobApplication {

  public static void main(String[] args) {
    SpringApplication.run(SjServiceJobApplication.class, args);
    LOGGER.info("======>>> JobServerApplication  Run Success");
  }


  @Bean(name="asyncExecutor")
  public Executor getAsyncExecutor() {
    ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
    exec.setMaxPoolSize(40);
    exec.setCorePoolSize(20);
    exec.setThreadGroupName("MyCustomExecutor");
    exec.setWaitForTasksToCompleteOnShutdown(true);
    exec.setBeanName("asyncExecutor");
    exec.initialize();
    return exec;
  }
}
5. 基于@Async无返回值调用

示例如下：

@Async  //标注使用
public void asyncMethodWithVoidReturnType() {
   System.out.println("Execute method asynchronously. "
     + Thread.currentThread().getName());
}
使用的方式非常简单，一个标注即可解决所有的问题。
6. 基于@Async有返回值的调用

示例如下：

@Async
public Future<String> asyncMethodWithReturnType() {
    System.out.println("Execute method asynchronously - "
      + Thread.currentThread().getName());
    try {
        Thread.sleep(5000);
        return new AsyncResult<String>("hello world !!!!");
    } catch (InterruptedException e) {
        //
    }
 
    return null;
}
以上示例可以发现，返回的数据类型为Future类型，其为一个接口。
具体的结果类型为AsyncResult,这个是需要注意的地方。

调用返回结果的异步方法示例：

public void testAsyncAnnotationForMethodsWithReturnType()
   throws InterruptedException, ExecutionException {
    System.out.println("Invoking an asynchronous method. "
      + Thread.currentThread().getName());
    Future<String> future = asyncAnnotationExample.asyncMethodWithReturnType();//调用自定义类asyncAnnotationExample的asyncMethodWithReturnType方法
 
    while (true) {  ///这里使用了循环判断，等待获取结果信息
        if (future.isDone()) {  //判断是否执行完毕
            System.out.println("Result from asynchronous process - " + future.get());
            break;
        }
        System.out.println("Continue doing something else. ");
        Thread.sleep(1000);
    }
}
分析： 这些获取异步方法的结果信息，是通过不停的检查Future的状态来获取当前的异步方法是否执行完毕来实现的。
7. 基于@Async调用中的异常处理机制

在异步方法中，如果出现异常，对于调用者caller而言，是无法感知的。如果确实需要进行异常处理，则按照如下步骤来进行处理：

1.  自定义实现AsyncTaskExecutor的任务执行器

//在这里定义处理具体异常的逻辑和方式。
public class ExceptionHandlingAsyncTaskExecutor implements AsyncTaskExecutor {
    private AsyncTaskExecutor executor;
    public ExceptionHandlingAsyncTaskExecutor(AsyncTaskExecutor executor) {
        this.executor = executor;
     }
      ////用独立的线程来包装，@Async其本质就是如此
    public void execute(Runnable task) {     
      executor.execute(createWrappedRunnable(task));
    }
    public void execute(Runnable task, long startTimeout) {
        //用独立的线程来包装，@Async其本质就是如此
       executor.execute(createWrappedRunnable(task), startTimeout);         
    } 
    public Future submit(Runnable task) { return executor.submit(createWrappedRunnable(task));
       //用独立的线程来包装，@Async其本质就是如此。
    } 
    public Future submit(final Callable task) {
      //用独立的线程来包装，@Async其本质就是如此。
       return executor.submit(createCallable(task)); 
    } 
    
    private Callable createCallable(final Callable task) { 
        return new Callable() { 
            public T call() throws Exception { 
                 try { 
                     return task.call(); 
                 } catch (Exception ex) { 
                     handle(ex); 
                     throw ex; 
                   } 
                 } 
        }; 
    }
    private Runnable createWrappedRunnable(final Runnable task) { 
         return new Runnable() { 
             public void run() { 
                 try {
                     task.run(); 
                  } catch (Exception ex) { 
                     handle(ex); 
                   } 
            }
        }; 
    } 
    private void handle(Exception ex) {
      //具体的异常逻辑处理的地方
      System.err.println("Error during @Async execution: " + ex);
    }
}
分析:可以发现其是实现了AsyncTaskExecutor, 用独立的线程来执行具体的每个方法操作。在createCallable和createWrapperRunnable中，定义了异常的处理方式和机制。
handle()就是未来我们需要关注的异常处理的地方。

2.  配置由自定义的TaskExecutor替代内置的任务执行器
<task:annotation-driven executor="exceptionHandlingTaskExecutor" scheduler="defaultTaskScheduler" />
<bean id="exceptionHandlingTaskExecutor" class="nl.jborsje.blog.examples.ExceptionHandlingAsyncTaskExecutor">
    <constructor-arg ref="defaultTaskExecutor" />
</bean>
<task:executor id="defaultTaskExecutor" pool-size="5" />
<task:scheduler id="defaultTaskScheduler" pool-size="1" />
8. @Async调用中的事务处理机制

在@Async标注的方法，同时也使用了@Transactional进行了标注；在其调用数据库操作之时，将无法产生事务管理的控制，原因就在于其是基于异步处理的操作。
那该如何给这些操作添加事务管理呢？可以将需要事务管理操作的方法放置到异步方法内部，在内部被调用的方法上添加@Transactional.

例如:

方法A，使用了@Async/@Transactional来标注，但是无法产生事务控制的目的。

方法B，使用了@Async来标注， B中调用了C、D，C/D分别使用@Transactional做了标注，则可实现事务控制的目的。



二.ThreadPoolExecutor的使用

线程资源必须线程池创建,通过ThreadPoolExecutor的方式,不建议在应用中自行显性地创建线程.
一方面是更加规范,合理的开辟和控制线程; 另一方面是把线程交给线程池管理可以优化资源效率.


虽然JDK中Executor框架有预定义线程池,但都有其局限性，不够灵活

ThreadPoolExecutor提供了四个构造方法：

new ThreadPoolExecutor(16, 16, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(9), new ThreadPoolExecutor.CallerRunsPolicy());

以上面这个实例化为例,其参数的释义:

参数名 类型 含义
corePoolSize int 核心线程池线程数量
maximumPoolSize int 线程池最大线程数量
keepAliveTime long 空闲线程存活时间
unit TimeUnit 存活时间单位
workQueue BlockingQueue<Runnable> 线程等待队列
threadFactory ThreadFactory 线程创建工厂
handler RejectedExecutionHandler 拒接执行策略

关键参数corePoolSize和maximumPoolSize的释义:

当提交新任务并且少于corePoolSize线程正在运行时，即使其他工作线程处于空闲状态，也会创建一个新线程来处理该请求。
如果当多于corePoolSize但小于maximumPoolSize时提交新任务，则仅当任务队列已满时才会创建新线程。

当maximumPoolSize已满时提交新任务会被RejectedExecutionHandler拒绝执行。

img



threadFactory参数的释义:

新线程使用ThreadFactory创建。 如果未另行指定，则使用Executors.defaultThreadFactory默认工厂，使其全部位于同一个ThreadGroup中，并且具有相同的NORM_PRIORITY优先级和非守护进程状态。
通过提供不同的ThreadFactory，您可以更改线程的名称，线程组，优先级，守护进程状态等。

workQueue参数的释义:

任务等待队列,一般分为以下几种:

  • 直接提交队列(SynchronousQueue)
  • 有界任务队列(ArrayBlockingQueue)
  • 无界任务队列(LinkedBlockingQueue)
  • 优先任务队列(PriorityBlockingQueue)

handler参数的释义:

如果出现任务队列已满且线程池创建的线程数达到你设置的最大线程数时，这时就需要你指定ThreadPoolExecutor的RejectedExecutionHandler参数即合理的拒绝策略，来处理线程池"超载"的情况。

ThreadPoolExecutor自带的拒绝策略如下：

  • AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作；
  • CallerRunsPolicy策略：如果线程池的线程数量达到上限，该策略会把任务队列中的任务放在调用者线程当中运行；
  • DiscardOledestPolicy策略：该策略会丢弃任务队列中最老的一个任务，也就是当前任务队列中最先被添加进去的，马上要被执行的那个任务，并尝试再次提交；
  • DiscardPolicy策略：该策略会默默丢弃无法处理的任务，不予任何处理。当然使用此策略，业务场景中需允许任务的丢失；
  • 也可以自定义策略

使用实例:

import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadTest {

    public static void main(String[] args) throws InterruptedException, IOException {
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);
        ThreadFactory threadFactory = new NameTreadFactory();
        RejectedExecutionHandler handler = new MyIgnorePolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                workQueue, threadFactory, handler);
        executor.prestartAllCoreThreads(); // 预启动所有核心线程

        for (int i = 1; i <= 10; i++) {
            MyTask task = new MyTask(String.valueOf(i));
            executor.execute(task);
        }
        
        
        /**
         submit()方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，
         submit方法还是调用的execute方法，只不过它利用了Future来获取任务执行结果。
         submit方法中参数可以是Callable类型也可以是Runnable类型，而execute()方法参数只能是Runnable类型。
         
         //调用了某service的getxXxxList()方法
         //Callable<Object> objectCallable = () -> xxxService.getxXxxList(id);
         //FutureTask<Object> objectCallableTask = new FutureTask<>(objectCallable);
         //executor.submit(objectCallableTask);
        */
        

    }
    
    //具体任务
    static class MyTask implements Runnable {
        private String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                System.out.println(this.toString() + " is running!");
                Thread.sleep(3000); //让任务执行慢点
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "MyTask [name=" + name + "]";
        }
    }

    //线程创建工厂
    static class NameTreadFactory implements ThreadFactory {

        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
            System.out.println(t.getName() + " has been created");
            return t;
        }
    }

    //任务拒绝策略
    public static class MyIgnorePolicy implements RejectedExecutionHandler {

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            doLog(r, e);
        }

        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志记录等
            System.err.println( r.toString() + " rejected");
           //System.out.println("completedTaskCount: " + e.getCompletedTaskCount());
        }
    }

}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 213,752评论 6赞 493
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,100评论 3赞 387
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 159,244评论 0赞 349
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,099评论 1赞 286
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,210评论 6赞 385
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,307评论 1赞 292
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,346评论 3赞 412
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,133评论 0赞 269
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,546评论 1赞 306
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,849评论 2赞 328
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,019评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,702评论 4赞 337
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,331评论 3赞 319
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,030评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,260评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,871评论 2赞 365
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,898评论 2赞 351