Java实现异步的几种方式

2023-12-14 17:17 由程长新发表于 #其他

普通线程实现异步，但频繁创建、销毁线程比较耗资源，所以一般交给线程池执行

//创建需要异步执行的逻辑
public class AsyncThread implements Runnable{
    @Override
    public void run() {
        System.out.println("异步线程开始");
        long start = System.currentTimeMillis();
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        long end = System.currentTimeMillis();
        System.out.println("异步线程：" + Thread.currentThread().getName() + "结束，耗时：" + (end - start));
    }
}

//在业务中进行调用
@GetMapping("/thread")
public String asyncThread(){
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 3, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.AbortPolicy());
    long start = System.currentTimeMillis();
    //自己的业务代码。。。
    AsyncThread asyncThread = new AsyncThread();
    threadPool.execute(asyncThread);
    long end = System.currentTimeMillis();
    return "返回,耗时:" + (end - start);
}

结果：

Future异步

和普通线程实现异步区别不大，只是使用Future是要获取执行后的返回值

//创建具有返回值的任务
public class CallableThread implements Callable {
    @Override
    public String call() throws Exception {
        long start = System.currentTimeMillis();
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        System.out.println("callable任务开始执行:" + start);
        TimeUnit.SECONDS.sleep(2);
        System.out.println();
        stopWatch.stop();
        System.out.println("stopWatch.prettyPrint------");
        System.out.println(stopWatch.prettyPrint());
        System.out.println("stopWatch.shortSummary------");
        System.out.println(stopWatch.shortSummary());
        System.out.println("stopWatch.getTotalTimeMillis------");
        System.out.println(stopWatch.getTotalTimeMillis());
        return "call执行结束 ";
    }
}

//在业务中进行调用
public String threadFuture(){
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 3, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.AbortPolicy());
    long start = System.currentTimeMillis();
    CallableThread callableThread = new CallableThread();
    Future<String> submit = threadPool.submit(callableThread);
    try {
        //在获取返回值时会阻塞主线程
        String s = "";
        s = submit.get();
        System.out.println(s);
    } catch (Exception e) {
        System.out.println("线程运行发生错误" + e.getMessage());
        throw new RuntimeException(e);
    }
    long end = System.currentTimeMillis();
    return "接口返回,耗时:" + (end - start);
}

结果：

Spring的@Async异步

使用@Async注解实现异步的前提是需要在启动类上标注@EnableAsync来开启异步配置

配置线程池（@Async默认情况下用的是SimpleAsyncTaskExecutor线程池，该线程池不是真正意义上的线程池，使用此线程池无法实现线程重用，每次调用都会新建一条线程。若系统中不断的创建线程，最终会导致系统占用内存过高，引发OutOfMemoryError错误）

/**
 * 线程池配置，可以配置多个线程池
 * @Async注解，默认使用系统自定义线程池，可在项目中设置多个线程池，在异步调用的时候，指明需要调用的线程池名称
 * 比如：@Async("线程池1")
 */
@Configuration
public class ExecutorConfig {
    /**
     * 自定义线程池
     */
    @Bean("myExecutor")
    public Executor taskExecutor(){
        System.out.println("系统最大线程数:" + Runtime.getRuntime().availableProcessors());
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(8);//核心线程数
        threadPoolTaskExecutor.setMaxPoolSize(16);//最大线程数
        threadPoolTaskExecutor.setQueueCapacity(1000);//配置队列容量
        threadPoolTaskExecutor.setKeepAliveSeconds(60);//空闲线程存活时间
        threadPoolTaskExecutor.setThreadNamePrefix("myExecutor-");//线程名字前缀
        return threadPoolTaskExecutor;
    }
}

编写异步方法的逻辑，异步方法所在的类需要被Spring管理

@Service
public class AsyncServiceImpl implements AsyncService {
    @Override
    @Async("myExecutor")
    public void sendMsg() {
        System.out.println("进入异步方法");
        System.out.println("当前线程名称：" + Thread.currentThread().getName());
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("异步方法执行完成");
    }

    /**
     * 具有返回值的异步方法，返回类型为Future,返回时new 一个AsyncResult对象，其中参数为返回的内容
     * @return
     */
    @Override
    @Async("myExecutor")
    public Future<String> sendMsgFuture() {
        System.out.println("进入future异步方法");
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return new AsyncResult<>("future异步方法执行完成");
    }
}

在业务逻辑中调用

@GetMapping("/asyncMethod")
public String asyncMethod(){
    System.out.println("aaa");
    System.out.println("调用异步方法");
    asyncService.sendMsg();
    System.out.println("bbb");
    return "asyncMethod方法返回";
}

调用没有返回值的异步方法结果：

@GetMapping("/asyncFutureMethod")
public String asyncFutureMethod(){
    System.out.println("aaa");
    Future<String> stringFuture = asyncService.sendMsgFuture();
    System.out.println("bbb");
    try {
    System.out.println(stringFuture.get());//get方法会阻塞主线程
    } catch (Exception e) {
    throw new RuntimeException(e);
    }
    return "asyncfutureMethod方法返回";
}

调用有返回值的异步方法结果：

Spring的ApplicationEvent事件实现异步

定义事件，继承ApplicationEvent类

public class MessageEvent extends ApplicationEvent {
    @Getter
    private String message;

    public MessageEvent(Object source, String message) {
        super(source);
        this.message = message;
    }
}

定义监听器（需要被Spring管理）

使用@EventListener注解写在方法上定义一个监听器，即事件被触发时执行的方法(默认是同步执行，可以使用@Async注解标注为异步执行)，支持多个监听器监听同一个事件。

@Component
public class MessageEventHandler {

    //@Async
    @EventListener
    public void handleLoginEvent(LoginEvent event){
        System.out.println("接受到LoginEvent事件");
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println(event.getUsername());
        System.out.println("LoginEvent事件处理完成");
    }

    //@Async
    @EventListener
    public void handleMessageEvent(MessageEvent event){
        System.out.println("接受到MessageEvent事件");
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println(event.getMessage());
        System.out.println("MessageEvent事件处理完成");
    }
}

定义事件发布者(触发事件的)（需要被Spring管理）

实现ApplicationEventPublisherAware接口

@Component
public class EventPublisher implements ApplicationEventPublisherAware {
    private ApplicationEventPublisher publisher;

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.publisher = applicationEventPublisher;
    }

    public void publish(ApplicationEvent event){
        if (event instanceof MessageEvent){
            System.out.println("开始发布MessageEvent事件:" + ((MessageEvent) event).getMessage());
        } else if (event instanceof LoginEvent) {
            System.out.println("开始发布LoginEvent事件:" + ((LoginEvent) event).getUsername());
        }
        //发布事件
        publisher.publishEvent(event);
        System.out.println("事件发布结束");
    }

}

业务代码执行时触发事件
```
@GetMapping("/pubEvent")
public String publishEvent(){
    System.out.println("业务逻辑开始");
    eventPublisher.publish(new MessageEvent(this,"testEvent"));
    System.out.println("业务逻辑结束");
    return "发布成功";
}
```
执行结果：

由控制台打印可以发现现在事件监听器方法的执行是同步的，如果需要异步执行，在监听器方法上加个@Async注解即可，但使用Async注解的前提是在启动类上标注@EnableAsync注解来开启异步配置

使用@Async注解后执行结果：

可以看到监听器中的打印在最后了，证明是异步执行的