关于ExecutorService 线程池的问题

0 0

关于ExecutorService 线程池的问题20

1.背景:
想要实现一个多线程扫描的功能,使用spring @Scheduled 设定扫描计划,每隔30s创建线程执行扫描任务.

2.出现的问题:
Scheduled 计划执行几天的时间后就会不再执行(方法里有日志输出,但不再打印),出现此问题后,系统还是可以正常访问,log中没有扫描日志

网上查询了一下,有人说是内存过低有可能导致Scheduled停止.

3.请问下面的代码有没有什么问题会导致内存泄露,关键代码已用红色标注

//相关引用

@Component
public class ScheduledScanTimer {
	private final Logger log = Logger.getLogger(getClass());
	private SimpleDateFormat sf = new SimpleDateFormat("HH:mm");
	private FiberScanTaskService scanTaskService;
	private FiberService fiberService;
	private FiberLogInfoService fiberLogInfoService;
	private FiberSDFIService fiberSDFIService;

	
	@Scheduled(fixedDelay = 30000)
	public void fiberScan() {
		log.info("@Scheduled  start......................");
		List<Future<List<ScanResult>>> scanList = new ArrayList<Future<List<ScanResult>>>();// 扫描结果
		// 整理扫描结果
		StringBuffer scanResultStr = new StringBuffer();
		StringBuffer fiberScanIDs = new StringBuffer();
		try {
			// 获取检查任务sql语句
			String sql = getScanSql();
			List<FiberScanTask> taskList = scanTaskService.getFiberScanTaskByScanTime(sql);
			// 检查是否有执行任务
			if (taskList != null && taskList.size() > 0) {
				log.info("have task  to execute!");
				for (FiberScanTask task : taskList) {
					// 设置任务状态为扫描中...
					task.setTaskStatus(1);
					scanTaskService.editFiberScanTask(task, false);
					// 获取光纤Ids
					String fiberIds = task.getFiberIds();
					// 对要扫描的光纤Ids按照sfdi分组
					List<ScanThread> scanThreadList = fiberService.getThreadGroupByFiberIds(fiberIds);
					// 初始化指定个数的线程池
					ExecutorService pool  = Executors.newFixedThreadPool(taskList.size());
					for (ScanThread st : scanThreadList) {
					    Callable<List<ScanResult>> callable = new FiberScanner(st.getFiberIds(), fiberService, fiberLogInfoService, fiberSDFIService);
					    Future<List<ScanResult>> future = pool.submit(callable);
					    scanList.add(future);
					}
					// 遍历线程返回结果
					for (Future<List<ScanResult>> fu : scanList) {
						List<ScanResult> resList = fu.get();
						for (ScanResult scanResult : resList) {
							scanResultStr.append(scanResult.getScanResultStr() + "<br>");
							fiberScanIDs.append(scanResult.getScanRecordId() + ",");
						}
					}
					// 关闭线程池
					pool.shutdown();
				}
			} else {
				log.info("no execute task !");
				// 更新无效任务状态
				updateInvalidTask();
			}
		} catch (InterruptedException e) {
			log.error("FiberController.java fiberScan()," + e.getMessage());
			e.printStackTrace();
		} catch (ExecutionException e) {
			log.error("FiberController.java fiberScan()," + e.getMessage());
			e.printStackTrace();
		} finally {
			scanList.clear();
			scanList = null;
			scanResultStr = fiberScanIDs = null;
		}
	}

}

如果换成 newCachedThreadPool 修改成下面的会不会好些?

//相关引用

@Component
public class ScheduledScanTimer {
	private final Logger log = Logger.getLogger(getClass());
	private SimpleDateFormat sf = new SimpleDateFormat("HH:mm");
	private FiberScanTaskService scanTaskService;
	private FiberService fiberService;
	private FiberLogInfoService fiberLogInfoService;
	private FiberSDFIService fiberSDFIService;
	private ExecutorService pool ;
	
	@Scheduled(fixedDelay = 30000)
	public void fiberScan() {
		log.info("@Scheduled  start......................");
		List<Future<List<ScanResult>>> scanList = new ArrayList<Future<List<ScanResult>>>();// 扫描结果
		// 整理扫描结果
		StringBuffer scanResultStr = new StringBuffer();
		StringBuffer fiberScanIDs = new StringBuffer();
		try {
			// 获取检查任务sql语句
			String sql = getScanSql();
			List<FiberScanTask> taskList = scanTaskService.getFiberScanTaskByScanTime(sql);
			// 检查是否有执行任务
			if (taskList != null && taskList.size() > 0) {
				log.info("have task  to execute!");
				// 初始化指定个数的线程池
				pool = Executors.newCachedThreadPool();
				for (FiberScanTask task : taskList) {
					// 设置任务状态为扫描中...
					task.setTaskStatus(1);
					scanTaskService.editFiberScanTask(task, false);
					// 获取光纤Ids
					String fiberIds = task.getFiberIds();
					// 对要扫描的光纤Ids按照sfdi分组
					List<ScanThread> scanThreadList = fiberService.getThreadGroupByFiberIds(fiberIds);
					for (ScanThread st : scanThreadList) {
					     Callable<List<ScanResult>> callable =new FiberScanner(st.getFiberIds(),fiberService,fiberLogInfoService,fiberSDFIService);
					     Future<List<ScanResult>> future = pool.submit(callable);
					     scanList.add(future);
					}
					// 遍历线程返回结果
					for (Future<List<ScanResult>> fu : scanList) {
						List<ScanResult> resList = fu.get();
						for (ScanResult scanResult : resList) {
							scanResultStr.append(scanResult.getScanResultStr() + "<br>");
							fiberScanIDs.append(scanResult.getScanRecordId() + ",");
						}
					}
				}
			} else {
				log.info("no execute task !");
				// 更新无效任务状态
				updateInvalidTask();
			}
		} catch (InterruptedException e) {
			// 关闭线程池
			pool.shutdown();
			log.error("FiberController.java fiberScan()," + e.getMessage());
			e.printStackTrace();
		} catch (ExecutionException e) {
			// 关闭线程池
			pool.shutdown();
			log.error("FiberController.java fiberScan()," + e.getMessage());
			e.printStackTrace();
		} finally {
			scanList.clear();
			scanList = null;
			scanResultStr = fiberScanIDs = null;
		}
	}

}

多线程 ExecutorService 线程池

2014年8月08日 09:35

gary_bu
10
0 0 3

从看代码上看，线程池的数量与每批任务数量相等，处理任务结果的方式是阻塞式的，
如果某一任务没有处理结束，那么本次定时器会一直阻塞，下次定时器执行时，若又出现类似情况，则又会出现定时器阻塞，
这样每次定时器执行时，线程池都得不到释放，则线程池会增加，线程也会越来越多，时间久了，就会出现楼主说的现象

个人觉得
　　在处理任务结果，可以使用带超时机制的get方法，
同时在定时器执行时，需要判断上一次是否执行完成了，否则不能开始新的扫描过程，不然就是找“死”的结果。

2014年8月12日 20:25