Spring 线程池使用
/**
*
*/
一个日志管理的类：
在批理更新，插入，等操作
package com.xiu.order.core.orders.impl;
import java.util.List;
import org.springframework.core.task.TaskExecutor;
import com.xiu.common.dao.persistence.exception.DAOException;
import com.xiu.order.core.exception.ManagerException;
import com.xiu.order.core.orders.LogManager;
import com.xiu.order.dal.orders.dao.LogDAO;
import com.xiu.order.dal.orders.dataobject.LogDO;
public class LogManagerImpl implements LogManager {
private LogDAO operatorLogDAO;
public LogDAO getOperatorLogDAO() {
return operatorLogDAO;
}
public void setOperatorLogDAO(LogDAO operatorLogDAO) {
this.operatorLogDAO = operatorLogDAO;
}
private TaskExecutor taskExecutor;
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@Override
public List<LogDO> queryOperatorLog(int orderId) throws ManagerException {
try {
return operatorLogDAO.queryOperatorLog(orderId);
} catch (DAOException e) {
throw new ManagerException(e);
}
}
@Override
public int insertOperatorLog(LogDO operationLogDO) throws ManagerException {
try {
return operatorLogDAO.insertOperatorLog(operationLogDO);
} catch (DAOException e) {
throw new ManagerException(e);
}
}
@Override
public List<LogDO> queryOperatorLogList(LogDO operationLogDO)
throws ManagerException {
try {
return operatorLogDAO.queryOperatorLogList(operationLogDO);
} catch (DAOException e) {
throw new ManagerException(e);
}
}
public void batchAddOptLogAsync(final List<LogDO> optLogDOList)
throws ManagerException {
taskExecutor.execute(new Runnable() {
public void run() {
try {
operatorLogDAO.batchAddOptLogAsync(optLogDOList);
} catch (DAOException e) {
// System.out.println(e);
}
}
});
}
@Override
public void addOptLogAsync(final LogDO optLogDO) throws ManagerException {
taskExecutor.execute(new Runnable() {
public void run() {
try {
insertOperatorLog(optLogDO);
} catch (ManagerException e) {
e.printStackTrace();
}
}
});
}
@Override
public List<Integer> queryLogInfoByOrderIdListAndLogInfo(
List<Integer> orderIdList, LogDO oprationLogDO)
throws ManagerException {
try {
return operatorLogDAO.queryLogInfoByOrderIdListAndLogInfo(
orderIdList, oprationLogDO);
} catch (DAOException e) {
throw new ManagerException(e);
}
}
}
spring 配置
<bean id="operatorLogManager" class="com.xiu.order.core.orders.impl.LogManagerImpl ">
<property name="taskExecutor" ref="taskExecutor"></property>
</bean>
<bean id="taskExecutor"
class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="10" />
<property name="maxPoolSize" value="30" />
</bean>
<转>
最近，由于项目里需要用到线程池来提高处理速度，记录一下spring的taskExecutor执行器来实现线程池。
这里省略了Service接口的定义和在applicationContext.xml文件中配置相应的bean（service）。
1、处理器实现类
Java代码
1.package com.shine.job;
2.
3.import java.util.List;
4.
5.import org.apache.commons.logging.Log;
6.import org.apache.commons.logging.LogFactory;
7.import org.springframework.core.task.TaskExecutor;
8.
9.import com.shine.service.Service;
10.
11./**
12. * 处理器接口实现类
13. */
14.public class ProcessorImpl implements Processor {
15.
16. // 日志对象
17. private static Log logger = LogFactory.getLog(ProcessorImpl.class);
18.
19. // 执行器
20. private TaskExecutor taskExecutor;
21.
22. // 业务接口
23. private Service service;
24.
25. public void process() {
26. // 使用多线程处理
27. taskExecutor.execute(new Runnable(){
28. public void run() {
29. try {
30. logger.debug("[Thread "
31. + Thread.currentThread().getId()
32. + " start]");
33. // 业务处理
34. service.handle(name);
35. logger.debug("[Thread "
36. + Thread.currentThread().getId()
37. + " end]");
38. } catch (RuntimeException e) {
39. logger.error("Service handle exception",e);
40. }
41. }
42. });
43. }
44. }
45.
46. public void setService(Service service) {
47. this.service = service;
48. }
49.
50. public void setTaskExecutor(TaskExecutor taskExecutor) {
51. this.taskExecutor = taskExecutor;
52. }
53.
54.}
2、applicationContext.xml配置
Xml代码
1.<!-- 线程池（执行器） -->
2.<task:executor id="taskExecutor" pool-size="1-4" queue-capacity="128" />
3.
4.<!-- 处理接口 -->
5.<bean id="processor" class="com.shine.job.ProcessorImpl">
6. <property name="service" ref="service" />
7. <property name="taskExecutor">
8. <ref bean="taskExecutor" />
9. </property>
10.</bean>
其中：
pool-size="1-4"，表示线程池活跃的线程数为1，最大线程数为4；
queue-capacity="128"，表示任务队列的最大容量。
PS：关于taskExecutor还有一种配置bean来实现的，其配置的写法和参数与上面基本一致。
ThreadPoolTaskExecutor的配置在网上找了很多解释没找到，看了下ThreadPoolExecutor的配置，名字差不多，应该含义也差不多。只不过ThreadPoolTaskExecutor对
ThreadPoolExecutor做了包装。
<bean id ="taskExecutor" class ="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" >
<property name ="corePoolSize" value ="5" />
<property name ="keepAliveSeconds" value ="300" />
<property name ="maxPoolSize" value ="10" />
<property name ="queueCapacity" value ="25" />
</bean> 线程的配置文件：
corePoolSize： 线程池维护线程的最少数量
keepAliveSeconds 线程池维护线程所允许的空闲时间
maxPoolSize 线程池维护线程的最大数量
queueCapacity 线程池所使用的缓冲队列
当一个任务通过execute(Runnable)方法欲添加到线程池时：
l 如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。
l 如果此时线程池中的数量等于 corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。
l 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。
l 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。也就是：处理任务的优先级为：核心线程corePoolSize、任务队列workQueue、最大线程 maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。
l 当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。
<!-- 云推送异步线程池 --> <bean id="pushThreadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <!-- 核心线程数 --> <property name="corePoolSize" value="5" /> <!-- 最大线程数 --> <property name="maxPoolSize" value="10" /> <!-- 队列最大长度 >=mainExecutor.maxSize --> <property name="queueCapacity" value="10000" /> <!-- 线程池维护线程所允许的空闲时间 --> <property name="keepAliveSeconds" value="300" /> <!-- 线程池对拒绝任务(无线程可用)的处理策略 --> <property name="rejectedExecutionHandler"> <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" /> </property> </bean>
转载请注明原文地址: https://ju.6miu.com/read-679306.html
