我想验证我编写的多线程应用程序的设计，并在几点上得到澄清/再保证。对于这么长的帖子，我提前道歉——我想把它分成几个问题，但是我必须引用相同的代码，而且它们似乎都是相互关联的，所以我选择把所有内容都放在一个帖子中。如果这不合适 - 请告诉我，我将把它分成多个帖子。
这是我所拥有的：
- BatchService（一个 Spring Singleton bean）：接受上传指定目录或 zip 存档的请求。为此，它拥有 ExecutorService servicePool 。在每个请求上，它都会向池提交一个新的 BatchUploader Callable 任务，并将返回的 Future 存储在一个列表中 - 一个 TX 方法。它提供了获取所有上传状态和取消所有上传的方法。它还启动一个新的 BatchMonitor 线程来监控上传进度并更新保存已完成和未完成上传信息的队列。它还会在 bean 即将被销毁时清理所有资源（使用 Spring 的 PreDestroy 回调）
- BatchUploader是一个 Callable 任务，它也有自己的 ServiceExecutor batchPool 来上传单个文件。在其 call() 方法中，它扫描目录或 zip 存档，并为每个文件提交一个 SingleFileUploader Callable 任务到其池。
- SingleFileUploader是一个 Callable 任务，在它的 call() 方法中，它完成所有上传和处理文件的工作并返回一些状态。
这是一些真实的和一些伪代码：
public class BatchService {
private ExecutorService servicePool;
private ConcurrentHashMap<String, Future<SingleBatchUploadResult>> uploadBatchFutures = new ConcurrentHashMap<String, Future<SingleBatchUploadResult>>();
// keep last 100 unsuccessful uploads
private ConcurrentLinkedQueue<SingleBatchUploadResult> notCompletedBatches = new ConcurrentLinkedQueue<SingleBatchUploadResult>();
// keep last 100 successful uploads
private ConcurrentLinkedQueue<String> completedBatches = new ConcurrentLinkedQueue<String>();
private Thread monitorThread;
public BatchService() {
executorService = Executors.newFixedThreadPool(MAX_BATCH_UPLOAD_THREADS);
monitorThread = new Thread(new BatchMonitor());
monitorThread.setDaemon(true);
monitorThread.start();
}
@Transactional
public void processUpload(String uploadId, String contentName) {
Future<SingleBatchUploadResult> taskFuture = servicePool.submit(new BatchUploader(uploadId, contentName));
uploadBatchFutures.put(uploadId, taskFuture);
}
@PreDestroy
public void preDestroy() {
// stop the monitor thread
monitorThread.interrupt();
// stop all executors and their threads
cancelAllTasks();
}
public void cancelAllTasks(){
List<Runnable> waitingTasks = servicePool.shutdownNow();
for (Runnable task: waitingTasks){
// examine which tasks are still waiting, if necessary
}
}
public boolean cancelBatchById(String uploadId){
Future<SingleBatchUploadResult> resultFuture = activeBatchFutures.get(uploadId);
if (resultFuture != null && (!resultFuture.isDone() || !resultFuture.isCancelled()) ){
resultFuture.cancel(true);
return true;
}
// this task was either already finished, cancelled, not submitted or unknown
return false;
}
public void getCurrentStatus(){
// just print out the sizes of queues for now
System.out.println("number of active uploads: " + activeBatchFutures.size());
System.out.println("number of successfully completed uploads: " + completedBatches.size());
System.out.println("number of failed uploads: " + notCompletedBatches.size());
}
public class BatchMonitor implements Runnable {
@Override
public void run() {
boolean cont = true;
while (cont) {
if (Thread.currentThread().isInterrupted()){
// the thread is being shut down - get out
cont = false;
break;
}
Iterator<Entry<String, Future<SingleBatchUploadResult>>> iterator = activeBatchFutures.entrySet().iterator();
// remove completed Futures from the map
// add successfully completed batches to completedBatches queue
// add all other batches to notCompletedBatches queue
while (iterator.hasNext() && cont){
…
if (batchUploadFuture.isCancelled()) {
addToNotCompleted(defaultResult);
// remove this future from the active list
activeBatchFutures.remove(uploadId);
} else if (batchUploadFuture.isDone()){
try {
SingleBatchUploadResult result = batchUploadFuture.get();
if (UploadStatus.SUCCESS.equals(result.getUploadStatus()))
addToCompleted(uploadId);
else
addToNotCompleted(result);
} catch (InterruptedException e) {
// the thread is being shut down - stop processing
cont = false;
// preserve interruption state of the thread
Thread.currentThread().interrupt();
break;
} catch (ExecutionException e) {
addToNotCompleted(defaultResult);
}
// remove this future from the active list
activeBatchFutures.remove(uploadId);
} else {
// the task has not finished yet - let it be
// TODO if a Future is not complete - see how old it is [how ?] If older then timeout - cancel it
// For now, rely on the ExecutorService timeout set on the BatchUploader
}
}
// try to sleep for 5 sec, unless the thread is being shutdown
if (!Thread.currentThread().isInterrupted()){
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
cont = false;
// preserve interruption state of the thread
Thread.currentThread().interrupt();
}
}
}
System.out.println("BatchMonitor.run() has terminated");
}
public void addToCompleted(String uploadId){
int currentSize = completedBatches.size();
// bring the size of the queue below MAX
if (currentSize > MAX_SUCCESSFUL_RESULTS) {
int delta = MAX_SUCCESSFUL_RESULTS - currentSize;
while (delta > 0){
completedBatches.poll();
delta--;
}
}
completedBatches.offer(uploadId);
}
public void addToNotCompleted(SingleBatchUploadResult result){
int currentSize = notCompletedBatches.size();
// bring the size of the queue below MAX
if (currentSize > MAX_UNSUCCESSFUL_RESULTS) {
int delta = MAX_UNSUCCESSFUL_RESULTS - currentSize;
while (delta > 0){
notCompletedBatches.poll();
delta--;
}
}
notCompletedBatches.offer(result);
}
}
}
public class BatchUploader implements Callable<SingleBatchUploadResult> {
private ExecutorService executorService;
// Map<fileName, Future result> - holds Futures for all files that were submitted for upload (those that did not fail validation)
private ConcurrentHashMap<String, Future<SingleFileUploadResult>> uploadTaskFutures = new ConcurrentHashMap<String, Future<SingleFileUploadResult>>();
private ConcurrentHashMap<String, SingleFileUploadResult> notUploadedFiles = new ConcurrentHashMap<String, SingleFileUploadResult>();
private int totalFilesToUpload = 0;
public BatchUploader(...) {
executorService = Executors.newFixedThreadPool(MAX_THREADS_PER_BATCH);
}
public SingleBatchUploadResult call() {
// do some validation
if ( this is a correct ZIP file){
String errorMessage = processZipArchive(threadName, contentName);
// the errorMessage will be not null if there were some exceptions that happened during the zip archive read:
// opening the ZIP archive, reading entries or thread interruption exceptions
if (errorMessage != null) {
...
return errorBatchUploadResult;
}
}
// all tasks are submitted - stop the service from accepting new requests and shutdown when done
executorService.shutdown();
// now wait until all tasks have finished - but only up to BATCH_UPLOAD_TIMEOUT_IN_SEC seconds
try {
executorService.awaitTermination(BATCH_UPLOAD_TIMEOUT_IN_SEC, TimeUnit.SECONDS);
} catch (InterruptedException e) {
// try to shutdown all running tasks and stop waiting tasks from being scheduled;
executorService.shutdownNow();
// preserve interruption state of the thread
Thread.currentThread().interrupt();
return errorBatchUploadResult;
}
// at this point, we either finished all tasks (awaitTermination finished before timeout),
// or we timed out waiting. Get the latest status of each task
List<String> successfullyUploadedFiles = new LinkedList<String>();
for (String entryName : uploadTaskFutures.keySet()) {
Future<SingleFileUploadResult> future = uploadTaskFutures.get(entryName);
try {
if (future.isCancelled()) {
...
notUploadedFiles.putIfAbsent(entryName, taskResult);
} else if (future.isDone()) {
// this task has finished
taskResult = future.get();
if (taskResult.getUploadStatus().equals(UploadStatus.SUCCESS))
successfullyUploadedFiles.add(entryName);
else
notUploadedFiles.putIfAbsent(entryName, taskResult);
} else {
// this task is either not started yet or not finished yet
…
notUploadedFiles.putIfAbsent(entryName, sometaskResult);
}
} catch (InterruptedException e){
// this is a signal to stop processing
batchUploadResult.setTotalFilesToUpload(totalFilesToUpload);
batchUploadResult.setNotUploadedFiles(notUploadedFiles);
batchUploadResult.setSuccessfullyUploadedFiles(successfullyUploadedFiles);
batchUploadResult.setStatusMessage(statusMessage);
batchUploadResult.setUploadStatus(UploadStatus.PARTIAL_FAILURE);
// cancel/stop all executing/waiting SingleFileUpload tasks
executorService.shutdownNow();
// preserve interruption state of the thread
Thread.currentThread().interrupt();
return batchUploadResult;
} catch (ExecutionException e) {
// we do not know what the state of this task is
…
notUploadedFiles.putIfAbsent(entryName, sometaskResult);
}
}
...
return batchUploadResult;
}
private String processZipArchive(String threadName, String zipName) {
// do all ZIP-reading work here
while ( valid file found )
{
if (Thread.currentThread().isInterrupted()){
// this batch uploader thread is being shut down - stop all SingleFileUpload tasks
executorService.shutdownNow();
return errorMessage;
}
// do a try while processing individual files to be able to gather info about failed files but continue processing good ones
try {
// read the file and pass it for processing to SingleFileUploader
Future<SingleFileUploadResult> taskFuture = executorService.submit(new SingleFileUploader(uploadId, bytesContent, zipEntryName));
uploadTaskFutures.put(zipEntryName, taskFuture);
...
} catch (some exceptions) {
notUploadedFiles.put(zipEntryName, taskResult);
}
}
return errorMessage;
}
}
public class SingleFileUploader implements Callable<SingleFileUploadResult> {
...
@Override
public SingleFileUploadResult call() {
// check if there was a cancellation request
if (Thread.currentThread().isInterrupted()){
// this file uploader thread is being shut down - get out
return errorResult;
}
// do the real work here
return result;
}
}
所有这些在常规情况下都可以正常工作。但是，我仍然想听听您对是否有更好/更可靠的方法来做我想做的事情的意见，尤其是在以下方面：
我正在使用一个单独的线程 BatchMonitor，通过定期扫描活动 Futures 列表并将它们移动到“成功完成”或“未完成 [失败]”队列来跟踪活动、已完成和尚未完成的内容。我想知道是否有更好的方法来做到这一点？
我为此使用同步的无界队列 - 并在我不断向它们添加项目时将它们绑定到指定的最大大小。我在标准 JDK 库中找不到“有界并发队列”，只有无界的，我希望我可以使用 Guava 的 EvictingQueue，但它被捆绑到似乎还没有发布的 15.0 版本中……所以，我决定自己限制队列的大小，但以使用 size() 操作为代价，我知道这是并发队列的问题，因为它会对队列进行全面扫描……我的理由是，如果我保持队列的大小很小 - 在我的情况下为 100
我需要并发队列吗？唯一修改队列的线程是 BatchMonitor 线程，而唯一将读取队列的其他线程是 BatchService 线程。唯一一次我可以进入不同步的情况是当 BatchSERvice 尝试获取特定上传的状态时。可能该上传已从 activeBatchFutures 映射中删除，但尚未放入“已完成”或“未完成”队列中，因为我不会故意在映射和队列之间同步读/写，以避免不必要的锁定。但是我可以离开，偶尔会为特定上传返回“未找到”状态 - 第二次询问状态会得到正确的结果。
BatchService 是一个 Singleton bean - 它带来了自己的可伸缩性问题，因为对该 bean 的所有请求都将受到限制。另一种选择是让每个 BatchUploader 成为 Spring bean 并限制 bean 的数量，但是我将如何进行整体监控？
处理超时和取消：我正在尝试使这个应用程序在资源清理方面防弹 - 我正在尝试处理所有 ThreadInterpution 案例并停止处理以允许线程被杀死。我依靠在 BAtchUploader 中捕获和处理的 InterruptedException，通过调用 batchPool.shutdownNow() 将此事件传播到各个 FileUploader 任务。你能看到我可能有失控线程的任何潜在情况 - 当 JVM 关闭时，应用程序重新部署在 Web 容器中，......？
谢谢！
码头