spring boot使用@Async注解解决异步多线程入库的问题

 更新时间:2022年05月27日 10:06:51   作者:jiuchengi  
最近在写项目是需要添加异步操作来提高效率,所以下面这篇文章主要给大家介绍了关于spring boot使用@Async注解解决异步多线程入库问题的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下

前言

在开发过程中,我们会遇到很多使用线程池的业务场景,例如定时任务使用的就是ScheduledThreadPoolExecutor。而有些时候使用线程池的场景就是会将一些可以进行异步操作的业务放在线程池中去完成,例如在生成订单的时候给用户发送短信,生成订单的结果不应该被发送短信的成功与否所左右,也就是说生成订单这个主操作是不依赖于发送短信这个操作,所以我们就可以把发送短信这个操作置为异步操作。而要想完成异步操作,一般使用的一个是消息服务器MQ,一个就是线程池。今天我们就来看看在Java中常用的Spring框架中如何去使用线程池来完成异步操作,以及分析背后的原理。

在Spring4中,Spring中引入了一个新的注解@Async,这个注解让我们在使用Spring完成异步操作变得非常方便。

在SpringBoot环境中,要使用@Async注解,我们需要先在启动类上加上@EnableAsync注解。这个与在SpringBoot中使用@Scheduled注解需要在启动类中加上@EnableScheduling是一样的道理(当然你使用古老的XML配置也是可以的,但是在SpringBoot环境中,建议的是全注解开发),具体原理下面会分析。加上@EnableAsync注解后,如果我们想在调用一个方法的时候开启一个新的线程开始异步操作,我们只需要在这个方法上加上@Async注解,当然前提是,这个方法所在的类必须在Spring环境中。

项目实况介绍

项目中,我需要将700w条数据,定时任务加入到mysql表中,去掉日志打印和一些其他因素的影响,入库时间还是需要8个小时以上,严重影响后续的一系列操作,所以我才用@Async注解,来实现异步入库,开了7个线程,入库时间缩短为1.5个小时,大大提高效率,以下是详细介绍,一级一些需要注意的坑.

需要写个配置文件两种方式

第一种方式

@Configuration
@EnableAsync //启用异步任务
public class ThreadConfig {
    @Bean
    public ThreadPoolTaskExecutor executor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
          //配置核心线程数
        executor.setCorePoolSize(15);
          //配置最大线程数
        executor.setMaxPoolSize(30);
          //配置队列大小
        executor.setQueueCapacity(1000);
          //线程的名称前缀
        executor.setThreadNamePrefix("Executor-");
          //线程活跃时间(秒)
        //executor.setKeepAliveSeconds(60);
          //等待所有任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
          //设置拒绝策略
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
          //执行初始化
        executor.initialize();
        return executor;
    }
}

第二种方式

@Configuration
@EnableAsync
public class ExecutorConfig {

   @Value("${thread.maxPoolSize}")
   private Integer maxPoolSize;
   @Value("${thread.corePoolSize}")
   private Integer corePoolSize;
   @Value("${thread.keepAliveSeconds}")
   private Integer keepAliveSeconds;
   @Value("${thread.queueCapacity}")
   private Integer queueCapacity;
   @Bean
   public ThreadPoolTaskExecutor asyncExecutor(){
      ThreadPoolTaskExecutor taskExecutor=new ThreadPoolTaskExecutor();
      taskExecutor.setCorePoolSize(corePoolSize);//核心数量
      taskExecutor.setMaxPoolSize(maxPoolSize);//最大数量
      taskExecutor.setQueueCapacity(queueCapacity);//队列
      taskExecutor.setKeepAliveSeconds(keepAliveSeconds);//存活时间
      taskExecutor.setWaitForTasksToCompleteOnShutdown(true);//设置等待任务完成后线程池再关闭
      taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//设置拒绝策略
      taskExecutor.initialize();//初始化
      return taskExecutor;
   }
}

配置文件

#线程池
thread:
  corePoolSize: 5
  maxPoolSize: 10
  queueCapacity: 100
  keepAliveSeconds: 3000

springboot默认是不开启异步注解功能的,所以,要让springboot中识别@Async,则必须在入口文件中,开启异步注解功能

package com.demo;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
 
//开启异步注解功能
@EnableAsync
@SpringBootApplication
public class SpringbootTaskApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(SpringbootTaskApplication.class, args);
    }
 
}

这里有个坑!

如果遇到报错:需要加上    proxyTargetClass = true

The bean 'xxxService' could not be injected as a'com.xxxx.xxx.xxxService' because it is a JDK dynamic proxy that implements:
xxxxxx
Action:
Consider injecting the bean as one of its interfaces orforcing the use of CGLib-based proxiesby setting proxyTargetClass=true on @EnableAsync and/or @EnableCaching.

当我service层处理完逻辑,吧list分成7个小list然后调用异步方法(异步方法的参数不用管,没影响,只截取核心代码)

List<List<DistributedPredictDTO>> partition = Lists.partition(userList, userList.size() / 7);
        for (List<DistributedPredictDTO> distributedPredictDTOS : partition) {
       //调用异步方法
            threadService.getI(beginDate, endDate, tableName, distributedPredictDTOS, hMap, i);
        }
@Slf4j
@Service
public class ThreadServiceImpl {
    @Resource
    ResourcePoolUrlProperties properties;
    @Resource
    private MonitorDao monitorDao;
    @Async
    Integer getI(String beginDate, String endDate, String tableName, List<DistributedPredictDTO> userList, Map<String, String> hMap, int i) {
        log.info("我开始执行");
        for (DistributedPredictDTO e : userList) {
            String responseStr;
            HashMap<String, String> pMap = Maps.newHashMap();
            pMap.put("scheduleId", e.getScheduleId());
            pMap.put("scheduleName", e.getScheduleName());
            pMap.put("distribsunStationId", e.getLabel());
            pMap.put("distribsunStationName", e.getValue());
            pMap.put("beginTime", beginDate);
            pMap.put("endTime", endDate);
            try {
                if ("180".equals(properties.getNewPowerSys().getDistributedPredictUrl().substring(17, 20))) {
                    pMap = null;
                }
                responseStr = HttpClientUtil.doPost(properties.getNewPowerSys().getDistributedPredictUrl(), hMap, pMap);
            } catch (Exception exception) {
                throw new RuntimeException(e.getValue() + "的功率预测接口异常" + hMap + pMap);
            }
            if (org.springframework.util.StringUtils.isEmpty(responseStr)) {
                log.info(e + "数据为空");
                continue;
            }
            JSONObject resJson = JSONObject.parseObject(responseStr);
            JSONObject obj = (JSONObject) resJson.get("obj");
            JSONArray tableData = (JSONArray) obj.get("tabledata");

            final List<DistributedUserPower> userPowers = Lists.newArrayList();
            for (Object o : tableData) {
                final DistributedUserPower distributedUserPower = new DistributedUserPower();
                distributedUserPower.setData(((JSONObject) o).get("data").toString());
                distributedUserPower.setData2(((JSONObject) o).get("data2").toString());
                distributedUserPower.setDataTime(((JSONObject) o).get("time").toString());
                distributedUserPower.setUserId(e.getLabel());
                distributedUserPower.setUserName(e.getValue());
                distributedUserPower.setAreaName(e.getScheduleName());
                distributedUserPower.setCreateTime(DateUtils.getDate());
                userPowers.add(distributedUserPower);
            }
            monitorDao.saveBatch(userPowers, tableName);
            i++;
        }
        return i;
    }

这里有两个坑!

第一个坑:

  我调用的异步方法在当前类中,则直接导致

@Async注解失效

正确操作,异步方法不要和同步调用方法写在同一个类中,应该重新调用其他类

第二个坑:

如果出现这个报错:

Null return value from advice does not mat

问题分析

代码中采用异步调用,AOP 做来一层切面处理,底层是通过 JDK 动态代理实现

不管采用 JDK 还是 CGLIB 代理,返回值必须是包装类型,所以才会导致上诉的报错信息

处理方案

将异步方法的返回值修改为基本类型的对应包装类型即可,如 int -> Integer

5分钟测试效果图:

最后一张是7线程:

总结

到此这篇关于spring boot使用@Async注解解决异步多线程入库问题的文章就介绍到这了,更多相关springboot @Async异步多线程入库内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • Java中如何对字符串进行utf-8编码

    Java中如何对字符串进行utf-8编码

    这篇文章主要介绍了Java中如何对字符串进行utf-8编码问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-04-04
  • Spring Cache的基本使用与实现原理详解

    Spring Cache的基本使用与实现原理详解

    缓存是实际工作中非经常常使用的一种提高性能的方法, 我们会在很多场景下来使用缓存。下面这篇文章主要给大家介绍了关于Spring Cache的基本使用与实现原理的相关资料,文中通过示例代码介绍的非常详细,需要的朋友可以参考下
    2018-05-05
  • SpringBoot监控Tomcat活动线程数来判断是否完成请求处理方式

    SpringBoot监控Tomcat活动线程数来判断是否完成请求处理方式

    这篇文章主要介绍了SpringBoot监控Tomcat活动线程数来判断是否完成请求处理方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-02-02
  • 详解SpringBoot程序启动时执行初始化代码

    详解SpringBoot程序启动时执行初始化代码

    这篇文章主要介绍了详解SpringBoot程序启动时执行初始化代码,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2018-09-09
  • java模式匹配之蛮力匹配

    java模式匹配之蛮力匹配

    这篇文章主要介绍了java模式匹配之蛮力匹配的相关资料和代码,需要的朋友可以参考下
    2015-05-05
  • java双色球机选法程序解析

    java双色球机选法程序解析

    这篇文章主要为大家详细解析了java双色球机选法程序,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2020-01-01
  • SpringBoot整合微信小程序支付V3(支付退款)

    SpringBoot整合微信小程序支付V3(支付退款)

    小程序支付在很多项目都会使用,本文主要介绍了SpringBoot整合微信小程序支付V3,具有一定的参考价值,感兴趣的可以了解一下
    2023-09-09
  • Java中List集合去重方法以及效率对比

    Java中List集合去重方法以及效率对比

    这篇文章主要给大家介绍了关于Java中List集合去重方法以及效率对比的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2021-04-04
  • Java实现优先队列式广度优先搜索算法的示例代码

    Java实现优先队列式广度优先搜索算法的示例代码

    这篇文章主要为大家详细介绍了Java如何实现优先队列式广度优先搜索算法,文中通过一个示例带大家具体了解了实现的方法,需要的可以参考一下
    2022-08-08
  • 详解SpringBoot如何开启异步编程

    详解SpringBoot如何开启异步编程

    本文主要介绍了详解SpringBoot如何开启异步编程,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2023-04-04

最新评论