我要评论背景
方法需要异步执行所以在需要的方法上添加了@async，启动类添加了@enableasync但碰到以下情况
1、注解无效
原因：我在调用的方法是写在同一个类下的，类似下面的所以不行
为此我将方法写在其他接口中引出了第二个问题
2、启动报错
原因找了好久是因为springboot是jdk动态代理，因为有实现类（imp）那必须得注入接口类（service层）通过接口类去引用实现的方法，但是还是报错原因居然是这样：
该实现层（imp）在其他地方有被直接注入（如
@autowired private doccheckserviceimpl doccheckserviceimpl;
那就不行了），所以其他地方改成接口类去实现其方法，启动就正常了
重点来了
如果你在实际项目开发中真这样写了，肯定会被老鸟们无情嘲讽
因为一般的代码忽略了一个最大的问题，就是给@async异步框架自定义线程池，
使用@async注解，在默认情况下用的是simpleasynctaskexecutor线程池，该线程池不是真正意义上的线程池。
使用此线程池无法实现线程重用，每次调用都会新建一条线程。若系统中不断的创建线程，最终会导致系统占用内存过高，引发outofmemoryerror错误。
正因如此，所以我们在使用spring中的@async异步框架时一定要自定义线程池，替代默认的simpleasynctaskexecutor。
spring提供了多种线程池：
simpleasynctaskexecutor：不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。synctaskexecutor：这个类没有实现异步调用，只是一个同步操作。只适用于不需要多线程的地concurrenttaskexecutor：executor的适配类，不推荐使用。如果threadpooltaskexecutor不满足要求时，才用考虑使用这个类threadpooltaskscheduler：可以使用cron表达式threadpooltaskexecutor：最常使用，推荐。其实质是对java.util.concurrent.threadpoolexecutor的包装
为@async实现一个自定义线程池：
@configuration
@enableasync
public class syncconfiguration {
@bean(name = "asyncpooltaskexecutor")
public threadpooltaskexecutor executor() {
threadpooltaskexecutor taskexecutor = new threadpooltaskexecutor();
//核心线程数
taskexecutor.setcorepoolsize(10);
//线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
taskexecutor.setmaxpoolsize(100);
//缓存队列
taskexecutor.setqueuecapacity(50);
//许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
taskexecutor.setkeepaliveseconds(200);
//异步方法内部线程名称
taskexecutor.setthreadnameprefix("async-");
/**
* 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumpoolsize，如果还有任务到来就会采取任务拒绝策略
* 通常有以下四种策略：
* threadpoolexecutor.abortpolicy:丢弃任务并抛出rejectedexecutionexception异常。
* threadpoolexecutor.discardpolicy：也是丢弃任务，但是不抛出异常。
* threadpoolexecutor.discardoldestpolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
* threadpoolexecutor.callerrunspolicy：重试添加当前的任务，自动重复调用 execute() 方法，直到成功
*/
taskexecutor.setrejectedexecutionhandler(new threadpoolexecutor.callerrunspolicy());
taskexecutor.initialize();
return taskexecutor;
}
}配置自定义线程池以后我们就可以大胆的使用@async提供的异步处理能力了。
为@async指定线程池名字
@sneakythrows
@async("asyncpooltaskexecutor")
public void dotask1() {
long t1 = system.currenttimemillis();
thread.sleep(2000);
long t2 = system.currenttimemillis();
log.info("task1 cost {} ms" , t2-t1);
}1、个人使用的示例1（等待全部执行完）
/**
* 异步更新值域
*/
executorservice executor = new threadpoolexecutor(
2,//核心线程数
5,//最大线程数
3,//线程空闲时间
timeunit.seconds,//时间单位
new linkedblockingdeque<>(3),
executors.defaultthreadfactory(),
new threadpoolexecutor.discardoldestpolicy());
@override
@async("asyncpooltaskexecutor")
public void updaterange() {
string lockkey = "rang_lock_key";
string lockvalue = "rang_lock_value";
final boolean lock = redislock.trylock(lockkey, lockvalue, checkconstant.check_lock_time);
if (!lock) {
throw new baseexception("值域更新进行中...请稍后再试!");
}
map<string, integer> countmap = new hashmap<>(16);
integer count = 0;
// 先查询rp_md_range 表中最新的时间
final list<string> maxtime = basemapper.selectmaxtime();
date lastupdatetime = new date();
boolean iseffectivetime = collectionutil.isnotempty(maxtime) && strutil.isnotblank(maxtime.get(0));
if (iseffectivetime) {
lastupdatetime = datetimeutil.formattime(maxtime.get(0));
}
long now = system.currenttimemillis();
log.info("值域开始更新值域：{}", now);
try {
date finallastupdatetime = lastupdatetime;
// 字典类型
completablefuture<void> dictfuture = completablefuture.runasync(() -> getdict(countmap, finallastupdatetime, iseffectivetime), executor);
// 药品
completablefuture<void> medicinefuture = completablefuture.runasync(() -> getmedicine(countmap, finallastupdatetime, iseffectivetime), executor);
// 医疗服务
completablefuture<void> serverfuture = completablefuture.runasync(() -> getserver(countmap, finallastupdatetime, iseffectivetime), executor);
// 手术
completablefuture<void> icdcmfuture = completablefuture.runasync(() -> geticdcm(countmap, finallastupdatetime, iseffectivetime), executor);
// 疾病
completablefuture<void> icdfuture = completablefuture.runasync(() -> geticd(countmap, finallastupdatetime, iseffectivetime), executor);
// 中医疾病
completablefuture<void> icdcnfuture = completablefuture.runasync(() -> geticdcn(countmap, finallastupdatetime, iseffectivetime), executor);
// 中医症候
completablefuture<void> icdcntypefuture = completablefuture.runasync(() -> geticdcntype(countmap, finallastupdatetime, iseffectivetime), executor);
// 等待全部执行完
completablefuture.allof(dictfuture, medicinefuture, serverfuture, icdcmfuture, icdfuture, icdcnfuture, icdcntypefuture).get();
for (string s : countmap.keyset()) {
count += countmap.get(s);
}
log.info("值域更新值域完成成! 耗时 {} 秒!", (system.currenttimemillis() - now) / 1000);
log.info("本次更新值域数量:{}条", count);
} catch (exception e) {
redislock.unlock(lockkey, lockvalue);
log.error("值域更新失败{}", e.getmessage());
throw new baseexception(e.getmessage());
} finally {
redislock.unlock(lockkey, lockvalue);
}
}2、个人示例2
/**
* 异步更新值域
*/
@override
@async("asyncpooltaskexecutor")
public void updaterange() {
list<mdrange> rangelist = new arraylist<>(16);
// 先查询rp_md_range 表中最新的时间
final list<string> maxtime = basemapper.selectmaxtime();
date lastupdatetime = new date();
boolean effectivetime = collectionutil.isnotempty(maxtime) && strutil.isnotblank(maxtime.get(0));
if (effectivetime) {
lastupdatetime = datetimeutil.formattime(maxtime.get(0));
}
try {
// 字典类型
date finallastupdatetime = lastupdatetime;
final list<dictdata> list = dictdataservice.list(new lambdaquerywrapper<dictdata>()
.eq(dictdata::getstatus, 0)
.gt(effectivetime, dictdata::getcreatetime, lastupdatetime)
.isnull(dictdata::getupdatetime)
.or(i -> i.eq(dictdata::getstatus, 0).gt(effectivetime, dictdata::getupdatetime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(dictdata::getdicttype, dictdata::getdictsort))
);
for (dictdata dictdata : list) {
mdrange mdrange = new mdrange();
mdrange.setkey(dictdata.getdictvalue());
mdrange.setvalue(dictdata.getdictlabel());
mdrange.settype(dictdata.getdicttype());
mdrange.setsort(sortenum.sort1.getkey());
rangelist.add(mdrange);
}
// 药品
final list<mdmedicinedata> medicinedatalist = medicinedataservice.list(new lambdaquerywrapper<mdmedicinedata>()
.eq(mdmedicinedata::getstatus, "1")
.gt(effectivetime, mdmedicinedata::getcreatetime, lastupdatetime)
.isnull(mdmedicinedata::getopertime)
.or(i -> i.eq(mdmedicinedata::getstatus, "1").gt(effectivetime, mdmedicinedata::getopertime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(mdmedicinedata::getmedicinename, mdmedicinedata::getcreatetime))
);
for (mdmedicinedata medicinedata : medicinedatalist) {
mdrange mdrange = new mdrange();
mdrange.setkey(medicinedata.getmedicinedataid());
mdrange.setvalue(medicinedata.getmedicinename());
mdrange.settype(sortenum.sort2.getdefaulttype());
mdrange.setsort(sortenum.sort2.getkey());
rangelist.add(mdrange);
}
// 医疗服务
final list<server> serverlist = serverservice.list(new lambdaquerywrapper<server>()
.eq(server::getstatus, "1")
.gt(effectivetime, server::getcreatetime, lastupdatetime)
.isnull(server::getopertime)
.or(i -> i.eq(server::getstatus, "1").gt(effectivetime, server::getopertime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(server::getcategoryid, server::getcreatetime))
);
for (server server : serverlist) {
mdrange mdrange = new mdrange();
mdrange.setkey(server.getservercode());
mdrange.setvalue(server.getservername());
mdrange.settype(sortenum.sort3.getdefaulttype());
mdrange.setsort(sortenum.sort3.getkey());
rangelist.add(mdrange);
}
// 手术
final list<icdcm> icdcmlist = icdcmservice.list(new lambdaquerywrapper<icdcm>()
.eq(icdcm::getstatus, "1")
.gt(effectivetime, icdcm::getcreatetime, lastupdatetime)
.isnull(icdcm::getopertime)
.or(i -> i.eq(icdcm::getstatus, "1").gt(effectivetime, icdcm::getopertime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(icdcm::getdatatype, icdcm::getcreatetime))
);
for (icdcm icdcm : icdcmlist) {
mdrange mdrange = new mdrange();
mdrange.setkey(icdcm.getcmcode());
mdrange.setvalue(icdcm.getcmname());
mdrange.settype(sortenum.sort5.getdefaulttype());
mdrange.setsort(sortenum.sort5.getkey());
rangelist.add(mdrange);
}
// 疾病
final list<icd> icdlist = icdservice.list(new lambdaquerywrapper<icd>()
.eq(icd::getstatus, "1")
.gt(effectivetime, icd::getcreatetime, lastupdatetime)
.isnull(icd::getopertime)
.or(i -> i.eq(icd::getstatus, "1").gt(effectivetime, icd::getopertime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(icd::getdatatype, icd::getcreatetime))
);
for (icd icd : icdlist) {
mdrange mdrange = new mdrange();
mdrange.setkey(icd.geticdcode());
mdrange.setvalue(icd.geticdname());
mdrange.settype(sortenum.sort4.getdefaulttype());
mdrange.setsort(sortenum.sort4.getkey());
rangelist.add(mdrange);
}
// 中医疾病
final list<icdcn> icdcnlist = icdcnservice.list(new lambdaquerywrapper<icdcn>()
.eq(icdcn::getstatus, "1")
.gt(effectivetime, icdcn::getcreatetime, lastupdatetime)
.isnull(icdcn::getopertime)
.or(i -> i.eq(icdcn::getstatus, "1").gt(effectivetime, icdcn::getopertime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(icdcn::getdatatype, icdcn::getcreatetime))
);
for (icdcn icdcn : icdcnlist) {
mdrange mdrange = new mdrange();
mdrange.setkey(icdcn.geticdcnid());
mdrange.setvalue(icdcn.geticdcnname());
mdrange.settype(sortenum.sort6.getdefaulttype());
mdrange.setsort(sortenum.sort6.getkey());
rangelist.add(mdrange);
}
// 中医症候
final list<icdcntype> icdcntypelist = icdcntypeservice.list(new lambdaquerywrapper<icdcntype>()
.eq(icdcntype::getstatus, "1")
.gt(effectivetime, icdcntype::getcreatetime, lastupdatetime)
.isnull(icdcntype::getopertime)
.or(i -> i.eq(icdcntype::getstatus, "1").gt(effectivetime, icdcntype::getopertime, finallastupdatetime))
.orderbyasc(true, arrays.aslist(icdcntype::getdatatype, icdcntype::getcreatetime))
);
for (icdcntype icdcntype : icdcntypelist) {
mdrange mdrange = new mdrange();
mdrange.setkey(icdcntype.geticdcnid());
mdrange.setvalue(icdcntype.geticdcnname());
mdrange.settype(sortenum.sort6.getdefaulttype());
mdrange.setsort(sortenum.sort6.getkey());
rangelist.add(mdrange);
}
} catch (exception e) {
log.info("值域更新失败{}", e.getmessage());
throw new baseexception(e.getmessage());
}
// 保存
if (rangelist.size() > 0) {
final boolean savebatch = savebatch(rangelist, rangelist.size());
if (savebatch) {
log.info("本次更新值域数量:{}条", rangelist.size());
}
} else {
log.info("本次无值域更新");
}
}3、个人示例3（不使用@async注解）
// 自定义线程池
@autowired
@qualifier("asyncpooltaskexecutor")
private threadpooltaskexecutor taskexecutor;
// 1、
final list<callable<list<dsstsettledailypagevo>>> tasklist = new arraylist<>(16);
// 2、
tasklist.add(() -> checkprocess(groupidstoremap, currentcheckdate, ledgervolist));
// 3、
list<future<list<dsstsettledailypagevo>>> futures = new arraylist<>(tasklist.size());
for (callable<list<dsstsettledailypagevo>> callable : tasklist) {
futures.add(taskexecutor.submit(callable));
}
for (future<list<dsstsettledailypagevo>> future : futures) {
dailypagevolist.addall(future.get());
}总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持代码网。
发表评论