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

1
2
3
4
5
6
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.0</version>
</dependency>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
<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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
//生产调度类
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();
}
}
}

1
2
3
4
5
6
7
8
9
10
11
public class consumeTask implements Runnable {
private String p;
public TransCodeTask(String p){
this.p=p;
}
@Override
public void run() {
System.out.println(p);
}
}