|
【郑州校区】学成在线-第19天-讲义-分布式事务 六
3.3 Spring Task并行任务
3.3.1 需求分析
在项目通常是需要多个不同的任务并行去执行。
本节实现Spring Task并行执行任务的方法。
3.3.2 配置异步任务
创建异步任务配置类，需要配置线程池实现多线程调度任务。
[AppleScript] 纯文本查看 复制代码 @Configuration
@EnableScheduling
public class AsyncTaskConfig implements SchedulingConfigurer, AsyncConfigurer {
//线程池线程数量
private int corePoolSize = 5;
@Bean
public ThreadPoolTaskScheduler taskScheduler()
{
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();//初始化线程池
scheduler.setPoolSize(corePoolSize);//线程池容量
return scheduler;
}
@Override
public Executor getAsyncExecutor() {
Executor executor = taskScheduler();
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return null;
}
@Override
public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
scheduledTaskRegistrar.setTaskScheduler(taskScheduler());
}
}
将@EnableScheduling添加到此配置类上，SpringBoot启动类上不用再添加@EnableScheduling
3.3.3 测试
通过测试发现两个任务由不同的线程在并行执行，互不影响。
4 订单服务定时发送消息
4.1 需求分析
定时任务发送消息流程如下：
1、每隔1分钟扫描一次任务表。
1、定时任务扫描task表，一次取出多个任务，取出超过1分钟未处理的任务
2、考虑订单服务可能集群部署，为避免重复发送任务使用乐观锁的方式每次从任务列表取出要处理的任务
3、任务发送完毕更新任务发送时间
关于任务表的添加：
正常的流程是订单支付成功向更新订单支付状态并向任务表写入“添加选课任务”。
目前订单支付功能没有开发，采用手动向任务表添加任务。
4.2 RabbitMQ配置
向RabbitMQ声明两个队列：添加选课、完成选课，交换机使用路由模式，代码如下：
[AppleScript] 纯文本查看 复制代码 @Configuration
public class RabbitMQConfig {
//添加选课任务交换机
public static final String EX_LEARNING_ADDCHOOSECOURSE = "ex_learning_addchoosecourse";
//添加选课消息队列
public static final String XC_LEARNING_ADDCHOOSECOURSE = "xc_learning_addchoosecourse";
//完成添加选课消息队列
public static final String XC_LEARNING_FINISHADDCHOOSECOURSE =
"xc_learning_finishaddchoosecourse";
//添加选课路由key
public static final String XC_LEARNING_ADDCHOOSECOURSE_KEY = "addchoosecourse";
//完成添加选课路由key
public static final String XC_LEARNING_FINISHADDCHOOSECOURSE_KEY = "finishaddchoosecourse";
/**
* 交换机配置
* @return the exchange
*/
@Bean(EX_LEARNING_ADDCHOOSECOURSE)
public Exchange EX_DECLARE() {
return
ExchangeBuilder.directExchange(EX_LEARNING_ADDCHOOSECOURSE).durable(true).build();
}
//声明队列
@Bean(XC_LEARNING_FINISHADDCHOOSECOURSE)
public Queue QUEUE_XC_LEARNING_FINISHADDCHOOSECOURSE() {
Queue queue = new Queue(XC_LEARNING_FINISHADDCHOOSECOURSE);
return queue;
}
//声明队列
@Bean(XC_LEARNING_ADDCHOOSECOURSE)
public Queue QUEUE_XC_LEARNING_ADDCHOOSECOURSE() {
Queue queue = new Queue(XC_LEARNING_ADDCHOOSECOURSE);
return queue;
}
/**
* 绑定队列到交换机 .
* @param queue the queue
* @param exchange the exchange
* @return the binding
*/
@Bean
public Binding
BINDING_QUEUE_FINISHADDCHOOSECOURSE(@Qualifier(XC_LEARNING_FINISHADDCHOOSECOURSE) Queue queue,
@Qualifier(EX_LEARNING_ADDCHOOSECOURSE) Exchange exchange) {
return
BindingBuilder.bind(queue).to(exchange).with(XC_LEARNING_FINISHADDCHOOSECOURSE_KEY).noargs();
}
@Bean
public Binding BINDING_QUEUE_ADDCHOOSECOURSE(@Qualifier(XC_LEARNING_ADDCHOOSECOURSE) Queue
queue, @Qualifier(EX_LEARNING_ADDCHOOSECOURSE) Exchange exchange) {
return
BindingBuilder.bind(queue).to(exchange).with(XC_LEARNING_ADDCHOOSECOURSE_KEY).noargs();
}
}
| 
收藏
淘帖
踩
