1

アプリケーションで1つのクラシックExecutorを使用しましょう。アプリケーションの多くの部分は、一部の計算にこのエグゼキュータを使用します。各計算はキャンセルできます。このために、shutdown()またはshutdownNow()エグゼキュータで呼び出すことができます。

しかし、Executorのタスクの一部のみをシャットダウンしたいと思います。悲しいことに、私はオブジェクトにアクセスできませんFuture。それらは計算実装のプライベートな部分です（実際には、計算はアクターフレームワークjetlangによってサポートされています）

Executorラッパーのようなものが必要です。これは、計算に渡すことができ、実際のExecutorによってサポートされる必要があります。このようなもの：

// main application executor
Executor applicationExecutor = Executors.newCachedThreadPool();

// starting computation
Executor computationExecutor = new ExecutorWrapper(applicationExecutor);
Computation computation = new Computation(computationExecutor);
computation.start();

// cancelling computation
computation.cancel();
// shutting down only computation tasks
computationExecutor.shutdown();

// applicationExecutor remains running and happy

または他のアイデア？

4

3 に答える 3

2

良い結果を望んでいる人のために：Ivan Sopovの答えに部分的に基づいた最終的な解決策があります。幸いなことに、jetlang はタスクのみを実行するためのExecutorインターフェイス (ではなくExecutorService) を使用するため、このラッパーによってのみ作成されたタスクの停止をサポートするラッパー クラスを作成します。

static class StoppableExecutor implements Executor {
    final ExecutorService executor;
    final List<Future<?>> futures = Lists.newArrayList();
    boolean stopped;

    public StoppableExecutor(ExecutorService executor) {
        this.executor = executor;
    }

    void stop() {
        this.stopped = true;
        synchronized (futures) {
            for (Iterator<Future<?>> iterator = futures.iterator(); iterator.hasNext();) {
                Future<?> future = iterator.next();
                if (!future.isDone() && !future.isCancelled()) {
                    System.out.println(future.cancel(true));
                }
            }
            futures.clear();
        }
    }

    @Override
    public void execute(Runnable command) {
        if (!stopped) {
            synchronized (futures) {
                Future<?> newFuture = executor.submit(command);
                for (Iterator<Future<?>> iterator = futures.iterator(); iterator.hasNext();) {
                    Future<?> future = iterator.next();
                    if (future.isDone() || future.isCancelled())
                        iterator.remove();
                }
                futures.add(newFuture);
            }
        }
    }
}

これを使用するのは非常に簡単です。

ExecutorService service = Executors.newFixedThreadPool(5);
StoppableExecutor executor = new StoppableExecutor(service);

// doing some actor stuff with executor instance
PoolFiberFactory factory = new PoolFiberFactory(executor);

// stopping tasks only created on executor instance
// executor service is happily running other tasks
executor.stop();

それで全部です。うまくいきます。

于 2011-11-04T17:07:57.287 に答える
0

Something like this? You may do partial shutdown:

for (Future<?> future : %ExecutorServiceWrapperInstance%.getFutures()) {
    if (%CONDITION%) {
        future.cancel(true);
    }
}

Here is the code:

package com.sopovs.moradanen;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ExecutorServiceWrapper implements ExecutorService {

private final ExecutorService realService;
private List<Future<?>> futures = new ArrayList<Future<?>>();

public ExecutorServiceWrapper(ExecutorService realService) {
    this.realService = realService;
}

@Override
public void execute(Runnable command) {
    realService.execute(command);
}

@Override
public void shutdown() {
    realService.shutdown();

}

@Override
public List<Runnable> shutdownNow() {
    return realService.shutdownNow();
}

@Override
public boolean isShutdown() {
    return realService.isShutdown();
}

@Override
public boolean isTerminated() {
    return realService.isTerminated();
}

@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    return realService.awaitTermination(timeout, unit);
}

@Override
public <T> Future<T> submit(Callable<T> task) {
    Future<T> future = realService.submit(task);
    synchronized (this) {
        futures.add(future);
    }
    return future;
}

public synchronized List<Future<?>> getFutures() {
    return Collections.unmodifiableList(futures);
}

@Override
public <T> Future<T> submit(Runnable task, T result) {
    Future<T> future = realService.submit(task, result);
    synchronized (this) {
        futures.add(future);
    }
    return future;
}

@Override
public Future<?> submit(Runnable task) {
    Future<?> future = realService.submit(task);
    synchronized (this) {
        futures.add(future);
    }
    return future;
}

@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
    List<Future<T>> future = realService.invokeAll(tasks);
    synchronized (this) {
        futures.addAll(future);
    }
    return future;
}

@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
        throws InterruptedException {
    List<Future<T>> future = realService.invokeAll(tasks, timeout, unit);
    synchronized (this) {
        futures.addAll(future);
    }
    return future;
}

@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
    //don't know what to do here. Maybe this method is not needed by the framework
    //than just throw new NotImplementedException();
    return realService.invokeAny(tasks);
}

@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
    //don't know what to do here. Maybe this method is not needed by the framework
    //than just throw new NotImplementedException();
    return realService.invokeAny(tasks, timeout, unit);
}
}
于 2011-11-03T11:27:10.297 に答える
0

ブール値のフラグが設定されるまで、(提供された を使用して実行する) にするのはComputationどうRunnableですか? Executor次の行に沿ったもの：

public class Computation
{
  boolean volatile stopped;

  public void run(){
    while(!stopped){
    //do magic
  }

  public void cancel)(){stopped=true;}
}

あなたがしていることは、本質的にスレッドを停止することです。ただし、ガベージ コレクションは行われず、Executor によって管理されるため、代わりに再利用されます。「スレッドを停止する適切な方法は何ですか?」を参照してください。

編集: 上記のコードは、while ループの本体が短時間かかることを想定しているという意味で非常に原始的であることに注意してください。そうでない場合、チェックは頻繁に実行されず、タスクをキャンセルしてから実際に停止するまでに遅延が発生します。

于 2011-11-03T10:45:36.230 に答える