本帖最后由 谷粒姐姐 于 2019-7-17 15:55 编辑
4.5 定时发送任务
4.5.1 Dao
添加更新任务方法：
[mw_shl_code=applescript,true]//更新任务处理时间 @Modifying @Query("update XcTask t set t.updateTime = :updateTime where t.id = :id ") public int updateTaskTime(@Param(value = "id") String id,@Param(value = "updateTime")Date updateTime);[/mw_shl_code]
4.5.2 Service
添加发送消息方法：
[mw_shl_code=applescript,true]/**
* //发送消息
* @param xcTask 任务对象
* @param ex 交换机id
* @param routingKey
*/ @Transactional public void publish(XcTask xcTask,String ex,String routingKey){
//查询任务
Optional<XcTask> taskOptional = xcTaskRepository.findById(taskId);
if(taskOptional.isPresent()){
XcTask xcTask = taskOptional.get();
//String exchange, String routingKey, Object object
rabbitTemplate.convertAndSend(ex,routingKey,xcTask);
//更新任务时间为当前时间
xcTask.setUpdateTime(new Date());
xcTaskRepository.save(xcTask);
} }[/mw_shl_code]
4.5.3 编写任务类
编写任务类，每分钟执行任务，启动订单工程，观察定时发送消息日志，观察rabbitMQ队列中是否有消息，代码 如下：
[mw_shl_code=applescript,true]package com.xuecheng.order.mq; @Component public class ChooseCourseTask { private static final Logger LOGGER = LoggerFactory.getLogger(ChooseCourseTask.class);
@Autowired
TaskService taskService;
//每隔1分钟扫描消息表，向mq发送消息
@Scheduled(fixedDelay = 60000)
public void sendChoosecourseTask(){
//取出当前时间1分钟之前的时间
Calendar calendar =new GregorianCalendar();
calendar.setTime(new Date());
calendar.add(GregorianCalendar.MINUTE,‐1);
Date time = calendar.getTime();
List<XcTask> taskList = taskService.findTaskList(time, 1000);
//遍历任务列表
for(XcTask xcTask:taskList){
//发送选课消息
taskService.publish(xcTask, xcTask.getMqExchange(),xcTask.getMqRoutingkey());
LOGGER.info("send choose course task id:{}",taskId);
}
} ｝[/mw_shl_code]
4.7 乐观锁取任务
考虑订单服务将来会集群部署，为了避免任务在1分钟内重复执行，这里使用乐观锁，实现思路如下：
1) 每次取任务时判断当前版本及任务id是否匹配，如果匹配则执行任务，如果不匹配则取消执行。
2) 如果当前版本和任务Id可以匹配到任务则更新当前版本加
1. 1、在Dao中增加校验当前版本及任务id的匹配方法
[mw_shl_code=applescript,true]public interface XcTaskRepository extends JpaRepository<XcTask, String> { //使用乐观锁方式校验任务id和版本号是否匹配，匹配则版本号加1
@Modifying
@Query("update XcTask t set t.version = :version+1 where t.id = :id and t.version = :version") public int updateTaskVersion(@Param(value = "id") String id,@Param(value = "version") int version);
...[/mw_shl_code]2、在service中增加方法，使用乐观锁方法校验任务
[mw_shl_code=applescript,true]@Transactional public int getTask(String taskId,int version){
int i = xcTaskRepository.updateTaskVersion(taskId, version);
return i; }
[/mw_shl_code]
3、执行任务类中修改
[mw_shl_code=applescript,true]...
//任务id String taskId = xcTask.getId();
//版本号 Integer version = xcTask.getVersion();
//调用乐观锁方法校验任务是否可以执行 if(taskService.getTask(taskId, version)>0){ //发送选课消息 taskService.publish(xcTask, xcTask.getMqExchange(),xcTask.getMqRoutingkey());
LOGGER.info("send choose course task id:{}",taskId);
} ...[/mw_shl_code]
|
|
收藏
淘帖
踩
