Application Events and Listeners
1、自定义事件和监听
1.1、定义事件
1 package com.cjs.boot.event; 2 3 import lombok.Data; 4 import org.springframework.context.ApplicationEvent; 5 6 @Data 7 public class BlackListEvent extends ApplicationEvent { 8 9 private String address; 10 11 public BlackListEvent(Object source, String address) { 12 super(source); 13 this.address = address; 14 } 15 }
1.2、定义监听
1 package com.cjs.boot.event; 2 3 import org.springframework.context.ApplicationListener; 4 import org.springframework.context.event.EventListener; 5 import org.springframework.stereotype.Component; 6 7 8 public class BlackListListener implements ApplicationListener<BlackListEvent> { 9 10 @Override 11 public void onApplicationEvent(BlackListEvent event) { 12 System.out.println("监听到BlackListEvent事件: " + event.getAddress()); 13 try { 14 Thread.sleep(2000); 15 } catch (InterruptedException e) { 16 e.printStackTrace(); 17 } 18 } 19 }
1.3、注册监听
1 package com.cjs.boot; 2 3 import com.cjs.boot.event.BlackListListener; 4 import org.springframework.boot.SpringApplication; 5 import org.springframework.boot.autoconfigure.SpringBootApplication; 6 import org.springframework.boot.web.server.ErrorPage; 7 import org.springframework.boot.web.server.ErrorPageRegistrar; 8 import org.springframework.boot.web.server.ErrorPageRegistry; 9 import org.springframework.cache.annotation.EnableCaching; 10 import org.springframework.context.annotation.Bean; 11 import org.springframework.http.HttpStatus; 12 import org.springframework.scheduling.annotation.EnableAsync; 13 14 @SpringBootApplication 15 public class CjsSpringbootExampleApplication { 16 17 public static void main(String[] args) { 18 19 SpringApplication springApplication = new SpringApplication(CjsSpringbootExampleApplication.class); 20 springApplication.addListeners(new BlackListListener()); 21 springApplication.run(args); 22 23 }
1.4、发布事件
1 package com.cjs.boot.controller; 2 3 import com.cjs.boot.event.BlackListEvent; 4 import org.springframework.beans.factory.annotation.Autowired; 5 import org.springframework.context.ApplicationContext; 6 import org.springframework.context.ApplicationEventPublisher; 7 import org.springframework.web.bind.annotation.GetMapping; 8 import org.springframework.web.bind.annotation.RequestMapping; 9 import org.springframework.web.bind.annotation.RestController; 10 11 @RestController 12 @RequestMapping("/activity") 13 public class ActivityController { 14 15 // @Autowired 16 // private ApplicationEventPublisher publisher; 17 18 @Autowired 19 private ApplicationContext publisher; 20 21 @GetMapping("/sayHello.json") 22 public void sayHello() { 23 24 /** 25 * You may register as many event listeners as you wish, but note that by default event listeners receive events synchronously. 26 * This means the publishEvent() method blocks until all listeners have finished processing the event. 27 */ 28 29 BlackListEvent event = new BlackListEvent(this, "abc@126.com"); 30 publisher.publishEvent(event); 31 System.out.println("事件发布成功"); 32 } 33 34 }
2、基于注解的事件监听
package com.cjs.boot.event; import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component; @Component public class BlackListListener { @EventListener public void processBlackListEvent(BlackListEvent event) { System.out.println(123); } } --- package com.cjs.boot; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class CjsSpringbootExampleApplication { public static void main(String[] args) { SpringApplication.run(CjsSpringbootExampleApplication.class, args); } }
3、异步监听
1 @EventListener 2 @Async 3 public void processBlackListEvent(BlackListEvent event) { 4 // BlackListEvent is processed in a separate thread 5 }
4、应用
6 import lombok.extern.slf4j.Slf4j; 7 import org.springframework.beans.factory.annotation.Autowired; 8 import org.springframework.context.event.EventListener; 9 import org.springframework.scheduling.annotation.Async; 10 import org.springframework.stereotype.Component; 11 12 import java.util.ArrayList; 13 import java.util.List; 14 import java.util.concurrent.ExecutionException; 15 import java.util.concurrent.Future; 16 import java.util.concurrent.atomic.AtomicInteger; 17 18 /** 19 * 批量送券 20 */ 21 @Slf4j 22 @Component 23 public class BatchSendCouponListener { 24 25 @Autowired 26 private CouponPresentLogService couponPresentLogService; 27 28 @Async 29 @EventListener 30 public void processBatchSendCouponEvent(BatchSendCouponEvent batchSendCouponEvent) { 31 Long cpId = batchSendCouponEvent.getCouponPresentId(); 32 log.info("收到BatchSendCouponEvent, cpId={}", cpId); 33 List<CouponPresentLogEntity> list = couponPresentLogService.selectByPid(cpId); 34 35 handle(cpId, list, 0); 36 } 37 38 private void handle(Long cpId, List<CouponPresentLogEntity> list, int times) { 39 if (times >= 2) { 40 log.info("超过重试次数退出, cpId: {}, 剩余: {}", cpId, list.size()); 41 return; 42 } 43 44 List<Future<CouponPresentLogEntity>> futureList = new ArrayList<>(); 45 46 for (CouponPresentLogEntity entity : list) { 47 futureList.add(couponPresentLogService.present(entity)); 48 } 49 50 AtomicInteger count = new AtomicInteger(0); 51 // 收集失败的 52 List<CouponPresentLogEntity> failList = new ArrayList<>(); 53 for (Future<CouponPresentLogEntity> future : futureList) { 54 try { 55 CouponPresentLogEntity couponPresentLogEntity = future.get(); 56 if (couponPresentLogEntity.getStatus() != PresentStatusEnum.SUCCESS.getType().intValue()) { 57 failList.add(couponPresentLogEntity); 58 } 59 count.getAndIncrement(); 60 if (count.intValue() >= list.size()) { 61 List<CouponPresentLogEntity> failPresentLogList = couponPresentLogService.selectFailLogByPid(cpId); 62 if (null != failPresentLogList && failPresentLogList.size() > 0) { 63 times++; 64 log.info("第{}次重试, CPID: {}, 总计: {}, 失败: {}", times, cpId, list.size(), failPresentLogList.size()); 65 handle(cpId, failPresentLogList, times); 66 } 67 } 68 } catch (InterruptedException e) { 69 log.error(e.getMessage(), e); 70 } catch (ExecutionException e) { 71 log.error(e.getMessage(), e); 72 } 73 } 74 } 75 76 }
1 import lombok.extern.slf4j.Slf4j; 2 import org.springframework.beans.factory.annotation.Autowired; 3 import org.springframework.scheduling.annotation.Async; 4 import org.springframework.scheduling.annotation.AsyncResult; 5 import org.springframework.stereotype.Service; 6 7 import javax.annotation.Resource; 8 import java.util.concurrent.*; 9 10 @Service 11 @Slf4j 12 public class CouponPresentLogServiceImpl implements CouponPresentLogService { 13 14 @Autowired 15 private CouponPresentLogDao couponPresentLogDao; 16 @Resource 17 private CouponSendRpcService couponSendRpcService; 18 19 @Async("myThreadPoolTaskExecutor") 20 @Override 21 public Future<CouponPresentLogEntity> present(CouponPresentLogEntity entity) { 22 try { 23 CouponBaseResponse rst = couponSendRpcService.send(entity.getUserId(), entity.getCouponBatchKey(), "1", entity.getVendorId()); 24 if (null != rst && rst.isSuccess()) { 25 entity.setStatus(PresentStatusEnum.SUCCESS.getType()); 26 entity.setFailureReason(PresentStatusEnum.SUCCESS.getName()); 27 }else { 28 String reason = (null == rst) ? "响应异常" : rst.getMsg(); 29 entity.setFailureReason(reason); 30 entity.setStatus(PresentStatusEnum.FAILURE.getType()); 31 } 32 }catch (Exception ex) { 33 log.error(ex.getMessage(), ex); 34 entity.setFailureReason(ex.getMessage()); 35 entity.setStatus(PresentStatusEnum.FAILURE.getType()); 36 } 37 couponPresentLogDao.update(entity); 38 39 return new AsyncResult<CouponPresentLogEntity>(entity); 40 } 41 42 }
5、统计异步任务执行的进度
利用Future获取执行结果，比如上面的例子中，由于不是直接提交的任务，所以用AsyncResult来返回结果
上面的例子中，一个大任务，然后下面有许多子任务。在主任务中，统计各子任务的执行情况，是成功还是失败，然后统计成功多少，失败多少
也可以这样写：
@Autowired ThreadPoolTaskExecutor taskExecutor; Future<Object> future = taskExecutor.submit(new Callable<Object>() { @Override public Object call() throws Exception { return null; } });