Java Multithreaded App - how to dynamically cancel Futures objects

Question

I think this is a common scenario for multithreaded Java applications so I'll try to describe it here.

In my Java App I've a threadExecutor object that defines a Pool of 5 Threads.

ExecutorService threadExecutor = Executors.newFixedThreadPool(5);

A sendCallables method is responsible to assign a List of Jobs to the Executor.
I keep track of a List with an ObjectX. In this way I can reference back the list of Futures if the User wants to interrupt/cancel the Threads. Something like this:

Map<ObjectX, List<Future<String>>> map = new HashMap<ObjectX, Future<String>>();

public void sendCallables(ObjectX referenceObj, List<Callable<String>> list) {
    ...
    List<Future<String>> futureList = new ArrayList<Future<String>>();
    for(Callable<String> myCallableJob : list) {
        Future<String> future = threadExecutor.submit(myCallableJob);
        futureList.add(future);
    }
    ...
    map.add(referenceObj, futureList);
    ...
}

public void cancelFutures(ObjectX referenceObj) {
    ...
    List<Future<String>> list = map.get(referenceObj);
    for(Future<String> future : list) {
        future.cancel();
    }
    map.remove(referenceObj);
    ....
}

So far so good.

Now there are cases when there's no need to execute the submitted tasks anymore.
In these situations the decision of cancelling the tasks should be taken intelligently/automatically by the Application.
Examples of such cases can be found in a Web Application when the User's session expires or when the specific Flow (related to the submitted tasks) ends before all the Jobs are executed.

So basically I'll need to call cancelFutures(referenceObj) everytime that there is no sense for my application to continue to execute the Jobs. I've to identify each situation when the application needs to call it.

I'm wondering if there is a better approach to do this.

I was thinking in a WeakHashMap to be able to cleanup the Map once the referenceObj is no more referenced by the Application but this does't stop the Futures to get executed as I still need to call .cancel() on them (a sort of eventHandler associated with WeakHashMap remove(Object) method ? )

Answer 1

If I correctly understand your problem, you have various types of activities happening and you don't want to tie these activities together just because they can be cancelled.

To me it sounds that each case that needs cancellation should trigger an event. Event listener would listen for these events and cancel relevant Future.

Your example about sessions - you'd have a class that implements javax.servlet.http.HttpSessionListner, detects relevant tasks that need cancellation and fires an event to cancel those tasks. As everything is asynchronous, you wouldn't care whether cancelled task is finished or not so nothing is lost.

Answer 2

I think the solution you are proposing is pretty good. If you really, really want to drive the cancellation from the referenceObj getting garbage collected then you can use a combination of WeakReference and ReferenceQueue.

Have one of these declared in your execution helper class.

ReferenceQueue<ObjectX> refQ = new ReferenceQueue<ObjectX>();

Do this each time a task batch is submitted

new WeakReference<ObjectX>( referenceObj, refQ ).enqueue();

Have a thread that just runs this loop which pulls off objects that have becomes weakly-reachable (are eligible for GC) and cancels the futures/tasks.

while (true)
{
    // this blocks
    ObjectX referenceObj = refQ.remove().get();
    cancelFutures( referenceObj );
}