0

我从另一个类中创建了这个线程，用于在完成时读取执行器的状态并在失败时取消其余任务。任务是可运行的

如果看到任何失败，整体状态必须为 1 或失败

final CompletionService completionService = new ExecutorCompletionService(getExecutorService());
final List<Future> futures = new ArrayList<Future>();

    FutureTask<Integer> tasks = new FutureTask<Integer>(new Callable<Integer>() {

        public Integer call() {

            int status = 0;
            boolean fail = false;

            try {
                for (int i = 0; i < 10; i++) {

                    MyRunnable resultObj = null;

                    try {
                        resultObj = (MyRunnable) completionService.take().get();
                    } catch (CancellationException e) {
                        // Skip it ..
                    }

                    if (!fail) {
                        status = resultObj.getStatus();

                        if (status == 1) {
                            fail = true;
                            for (Future future : futures) {
                                if (!future.isCancelled() && !future.isDone())
                                    future.cancel(true); // cancel pending tasks including running tasks 
                            }
                        }
                    }
                }
            } catch (Exception e) {
                 e.printStackTrace();
            }

            return status;
        }

            });

上面的线程已启动-

ExecutorService pool = Executors.newSingleThreadExecutor();
pool.submit(tasks);

在下面，对象是从池中借用的，这是一个阻塞调用，我将池大小设置为 3 所以最初立即创建了 3 个 MyRunnable 工作人员。当每个工人完成时，他们会重复使用以服务于其余的任务。

for (int i = 0 ; i < 10; i ++;) {

    MyRunnable myRunnable = null;
    myRunnable = (MyRunnable) this.getGenericObjectPool().borrowObject();

    set myRunnable ..

    futures.add(completionService.submit(myRunnable, myRunnable));

}

while (!tasks.isDone()) {

        try {
            Thread.sleep(Global.WaitTime());            
        } catch (InterruptedException iex) {            
        }

}

finalStatus = tasks.get();
pool.shutdown();

GenericObjectPool 被配置为重用对象。我通过强制第一个线程失败并将其状态设置为 1 在 IDE 中模拟了一个测试。但是，问题是一旦它被释放，它就被 borrowObject() 重用，并且监控线程看到了状态已设置的更改对象作为由 GenricObjectPool 完成的新对象激活的一部分，返回 0。

所以，我无法从失败的线程中读取状态。MyRunnable 不是 Callable ，所以我不得不使用 completionService.submit(obj,obj) 来欺骗 Runnable

如果将池大小设置为 10 或更大，则不会发生此问题，因为这样就不会重用任何对象，并且我将成功读取每个对象的状态，但这不是一个选项。

4

1 回答 1

0

我为 Runnable 创建了一个 CallableDecorator 来解决这个问题。现在，即使使用 GenericObjectPool，我也有正确的返回值。由于现在不依赖于 Pool 对象来读取状态，即使重用对象也不会导致状态重置 -

因此，代码中的 2 处更改 - 更改

futures.add(completionService.submit(myRunnable, myRunnable));

到

futures.add(completionService.submit(new CallableDecorator(myRunnable)));

添加一个新类

public class CallableDecorator implements Callable {

       IRunnable r;

       public CallableDecorator(IRunnable r) {

           this.r = r;
       }

       public Integer call() {

           r.run();
           return r.statusCode();
       }
}

interface IRunnable extends Runnable {
     public Integer statusCode();
}

同样，必须将 resultObj 更改为整数才能在监视器线程中获取其值。

于 2013-02-28T14:45:56.870 回答