quartz+线程安全队列+ExecutorService线程池实现生产消费模型

75 阅读 0 评论 50 点赞

我是靠谱客的博主开朗宝马，最近开发中收集的这篇文章主要介绍quartz+线程安全队列+ExecutorService线程池实现生产消费模型，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

环境

1.maven
2.java8
3.eclipse

涉及到的组件框架

1.springMVC
2.quartz2.3.0
3.ConcurrentLinkedQueue
4.ExecutorService

进入正题

1.配置pom文件，加入quartz依赖包，maven项目下载依赖构建项目

<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.0</version>
</dependency>

2.在xml中加入quartz的spring配置

<bean id="producter" class="com.uws.schedule.jobs.ProduceSchedule"></bean>
<!-- 定义调用对象和调用对象的方法 -->
<bean id="JobWork"
class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean">
<!-- 调用的类 -->
<property name="targetObject" ref="MyJob"></property>
<!-- 调用类中的方法 -->
<property name="targetMethod">
<value>execute</value>
</property>
</bean>
<!-- 定义触发时间 -->
<bean id="JobTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
<property name="jobDetail" ref="JobWork" />
<!-- 每隔5秒执行一次 -->
<property name="cronExpression" value="*/5 * * * * ?" />
</bean>
<bean id="JobSchedule" lazy-init="false" autowire="no"
class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="triggers">
<list>
<ref bean="JobTrigger" />
</list>
</property>
</bean>

实现生产消费有多种方法，下面介绍其中一种

一对一消费，也就是一个消费线程只消费一个产品
1.创建生产调度类ProduceSchedule

//生产调度类
public class ProduceSchedule{
//线程池ThreadPoolExecutor参数说明
//(最小初始线程数，最大线程数，线程空闲清理时间，时间单位，线程队列/有阻塞和非阻塞之分，线程工厂，拒绝模式当超出线程数量限制时如何拒绝)
private static ExecutorService pool = new ThreadPoolExecutor(3, 5, 1000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
//线程安全的任务队列，不允许null元素入列
private static ConcurrentLinkedQueue<String> concurrentLinkedQueue = new ConcurrentLinkedQueue<String>();
public void execute(){
//先判断任务队列是否为空，不为空证明处理线程还活着，为空证明处理线程已经挂起需要从新激活线程池中的处理线程
boolean sw=concurrentLinkedQueue.isEmpty();
try {
List<String> ls=getProducts();//获取生产产品列表，getProducts()业务方法，自行实现
if(ls.size()>0){//发现有新的产品被生产
for(String s:ls){//获取新生产的产品
concurrentLinkedQueue.add(s);//将产品加入队列
}
}
if(sw){//为true证明之前任务是闲置的，需要再次执行while循环处理线程
//如果false，证明队列中仍然有任务，所以while循环还在继续执行，就无需再次执行
while(!concurrentLinkedQueue.isEmpty()){
//从线程池中获取一个消费线程，执行消费任务
pool.execute(new ConsumeTask(concurrentLinkedQueue.poll()));
}
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

2.消费线程类consumeTask

public class consumeTask implements Runnable {
private String p;
public TransCodeTask(String p){
this.p=p;
}
@Override
public void run() {
System.out.println(p);
}
}

其他的还有固定消费线程消费，比如3个消费者，每个消费者消费多个产品，需要进行线程监控，比如初始创建3个消费线程，将任务队列作为参数传入消费线程，只要任务队列中有任务，则三个消费线程就会一直从任务队列中领取产品消费，直到队列为空，当有新的任务进入队列时，需要先判断三个线程是否挂起状态，挂起的激活，运行状态的就继续执行即可。