依赖包
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-bus</artifactId> <version>2.0.7.RELEASE</version> </dependency> <dependency> <groupId>io.projectreactor.spring</groupId> <artifactId>reactor-spring-core</artifactId> <version>2.0.7.RELEASE</version> </dependency> <dependency> <groupId>io.projectreactor.spring</groupId> <artifactId>reactor-spring-context</artifactId> <version>2.0.7.RELEASE</version> </dependency>
类BaseEvent
@Component public class BaseEvent implements ApplicationContextAware{ private static ApplicationContext context; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { context = applicationContext; } public static EventBus getEventBus(){ return context.getBean(EventBus.class); } }
TestSerivce
@Service @Consumer public class TestSerivce extends BaseEvent{ private static final Log log = LogFactory.getLog("test"); @Autowired private CommonDao dao; @Autowired private TestBorDao; @Autowired private ThreadPoolTaskExecutor executor; final static String EVT_CLEAR_COMPLETED = "EVT_CLEAR_COMPLETED"; final static String EVT_UPDATE_COMPLETED = "EVT_UPDATE_COMPLETED"; final static String EVT_START_CLEAN_UP = "EVT_START_CLEAN_UP"; public void checkMain(){ //当前系统存在问题,有2层事务sprint和hibernate,导致无法手动控制事务,必须异步跳出,使内外执行处于不同事物中 log.info("开始贷后扫描"); Future<Boolean> result = executor.submit(new Callable<Boolean>() { @Override public Boolean call() throws Exception { try { //检查已监控但未启用的数据,将列表中旧检查数据置为已删除,并物理删除半年以前的数据 checkNotEnabled(); } catch (Exception e) { log.error("检查已监控但未启用的数据出错",e); } try { //检查未监控但已启用的数据,将其加入贷后监控主表 checkEnabled(); } catch (Exception e) { log.error("检查未监控但已启用的数据出错",e); } return true; } }); try{ result.get(); getEventBus().notify(EVT_CLEAR_COMPLETED); }catch(Exception e){ } } @Selector(EVT_CLEAR_COMPLETED) public void update(){ final StringBuffer errorMsg=new StringBuffer(); for (final TestMain mo : dao.find(TestMain.class,"isDeleted = 1")){ Future<Boolean> result = executor.submit(new Callable<Boolean>() { @Override public Boolean call() throws Exception { try { updateData(mo); return true; } catch (Exception e) { log.error("数据处理失败,ID"+mo.getId(),e); errorMsg.append("数据处理失败,ID"+mo.getId()); return false; } } }); try{ if (!result.get()) mo.setIsDeleted(0); getEventBus().notify(EVT_UPDATE_COMPLETED,Event.wrap(mo)); }catch(Exception e){ getEventBus().notify(EVT_UPDATE_COMPLETED,Event.wrap(mo)); } } getEventBus().notify(EVT_START_CLEAN_UP,Event.wrap(errorMsg.toString())); } @Selector(EVT_UPDATE_COMPLETED) public void update(Event<RkMonMain> mo){ try{ dao.saveOrUpdate(mo.getData()); }catch(Exception e){ log.error(e); } } @Selector(EVT_START_CLEAN_UP) public void cleanup(Event<String> result){ String errors = result.getData(); if(!CommValidation.isEmpty(errors)){ List<SysAutowarnMember> amList=dao.find(SysAutowarnMember.class,"isDeleted = 1 "); String toMail=""; for(SysAutowarnMember member:amList){ toMail+=member.getEmail()+","; } log.warn("跑批没有全部成功:"+errors); messageService.sendMessage("数据处理失败,ID",errors,toMail,null,null); } } private void updateData(TestMain mo){ } }
通过通知的方式实现异步执行 (编辑:李大同)
【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容!
|