依赖包
<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){
}
}
通过通知的方式实现异步执行
