java如何使外部方法可中断？

Question

问题

我正在通过ExecutorService运行一些外部方法的多次调用。我希望能够中断这些方法，但不幸的是，它们自己并没有检查中断标志。有没有办法强制从这些方法引发异常

我知道，从任意位置抛出异常是potentially dangerous，在我的具体情况下，我愿意抓住这个机会，并准备处理后果

细节

我所说的“外部方法”是指一些来自外部库的方法，我不能修改它的代码（我可以，但只要发布新版本，它就会成为维护的噩梦）

外部方法的计算成本很高，不受IO限制，因此它们不会对常规中断做出响应，我也无法强制关闭通道、socket或其他东西。正如我前面提到的，它们也不检查中断标志

该代码在概念上类似于：

// my code
public void myMethod() {
    Object o = externalMethod(x);
}

// External code
public class ExternalLibrary {
    public Object externalMethod(Object) {
        innerMethod1();
        innerMethod1();
        innerMethod1();
    }

    private void innerMethod1() {
        innerMethod2();
        // computationally intensive operations
    }

    private void innerMethod2() {
        // computationally intensive operations
    }
}

我试过的

^{}理论上可以做我想做的事情，但它不仅已被弃用，而且仅适用于实际的线程，而我正在处理executor任务（也可能与将来的任务共享线程，例如在线程池中工作时）。然而，如果找不到更好的解决方案，我会将代码转换为使用老式线程，并使用此方法

我尝试过的另一个选择是用一个特殊的“可中断”注释标记myMethod()和类似的方法，然后使用AspectJ（我承认是新手）捕获那里的所有方法调用，比如：

@Before("call(* *.*(..)) && withincode(@Interruptable * *.*(..))")
public void checkInterrupt(JoinPoint thisJoinPoint) {
    if (Thread.interrupted()) throw new ForcefulInterruption();
}

但是withincode对匹配方法调用的方法不是递归的，因此我必须将此注释编辑到外部代码中

最后，这与a previous question of mine类似——尽管一个显著的区别是，现在我处理的是一个外部库

Answer 1

我想到了以下奇怪的想法：

• 使用字节码修改库，如Javassist，在字节码的各个点引入中断检查。仅仅在方法的开头可能是不够的，因为您提到这些外部方法不是递归的，所以您可能希望在任何时候强制停止它们。在字节码级别执行此操作也将使其具有非常高的响应性，例如，即使外部代码在循环中运行，也可能引入中断检查。但是，这会增加一些开销，因此总体性能会变慢
• 为外部代码启动单独的进程（例如单独的VM）。中止进程可能比其他解决方案更容易编码。缺点是您需要外部代码和您的代码之间的某种通信通道，例如IPC、套接字等。第二个缺点是您需要更多的资源（CPU、内存）来启动新的虚拟机，并且可能是特定于环境的。如果您使用外部代码启动几个任务，而不是数百个任务，那么这将起作用。此外，性能也会受到影响，但计算本身将与原始计算一样快。可以使用java强制停止进程。lang.Process。销毁
• 使用自定义安全管理器，该管理器对每个checkXXX方法执行中断检查。如果外部代码以某种方式调用特权方法，您可以在这些位置中止。java就是一个例子。lang.SecurityManager。如果外部代码定期读取系统属性，请检查PropertyAccess（字符串）

Answer 2

你写道：

Another option I've tried is to mark myMethod() and similar methods with a special "Interruptable" annotation and then use AspectJ (which I am admittedly a newbie at) for catching all method invocations there - something like:

@Before("call(* *.*(..)) && withincode(@Interruptable * *.*(..))")
public void checkInterrupt(JoinPoint thisJoinPoint) {
    if (Thread.interrupted()) throw new ForcefulInterruption();
}

But withincode isn't recursive to methods called by the matching methods, so I would have to edit this annotation into the external code.

AspectJ的想法很好，但您需要

• 使用cflow()或cflowbelow()递归地匹配某个控制流（例如@Before("cflow(execution(@Interruptable * *(..)))")）
• 确保也编织您的外部库，而不仅仅是您自己的代码。这可以通过使用二进制编织、检测JAR文件的类并将它们重新打包到新的JAR文件中来实现，也可以通过在应用程序启动期间（即在类加载期间）应用LTW（加载时编织）来实现

如果您的外部库有一个包名，您甚至可能不需要标记注释，您可以用within()来确定它。AspectJ非常强大，解决问题的方法通常不止一种。我建议使用它，因为它是为你这样的努力而制作的

Answer 3

如果内部方法具有类似的名称，则可以使用xml（spring/AspectJ）中的切入点定义而不是注释，因此不需要修改外部库的代码

Answer 4

我想出了一个解决我问题的丑陋的办法。这并不漂亮，但它在我的情况下起作用，所以我把它贴在这里，以防它会帮助其他人

我所做的是分析我的应用程序的库部分，希望我能分离出一小组被重复调用的方法，例如一些get方法或equals()或类似的方法；然后我可以在那里插入以下代码段：

if (Thread.interrupted()) {
    // Not really necessary, but could help if the library does check it itself in some other place:
    Thread.currentThread().interrupt();
    // Wrapping the checked InterruptedException because the signature doesn't declare it:
    throw new RuntimeException(new InterruptedException());
}

可以通过编辑库的代码手动插入，也可以通过编写适当的方面自动插入。请注意，如果库试图捕获并吞下RuntimeException，抛出的异常可能会被库不尝试捕获的其他异常所替换

幸运的是，使用VisualVM，我能够找到一个单个方法，在我对库的特定使用过程中调用了很多次。添加上述代码段后，它现在可以正确响应中断

这当然是不可维护的，再加上没有什么能真正保证库在其他场景中会重复调用此方法；但它对我起作用，而且因为比较容易地对其他应用程序进行配置并在其中插入检查，所以我认为这是一个通用的，如果是丑陋的解决方案。p>

Answer 5

这个解决方案也不容易，但它可以工作：使用Javassist或CGLIB，您可以在每个内部方法（可能是由main run（）方法调用的方法）的开头插入代码，以检查线程是否处于活动状态，或者其他标志（如果是其他标志，您还必须添加它，以及设置它的方法）

我建议使用Javassist/CGLIB，而不是通过代码扩展类，因为您提到它是外部的，您不想更改源代码，而且将来可能会更改。因此，在运行时添加中断检查将适用于当前版本，也适用于将来的版本，即使内部方法名称发生更改（或其参数、返回值等）。您只需获取该类并在每个非run（）方法的方法的开头添加中断检查

Answer 6

一种选择是：

1. Make the VM connect to itself using JDI.
2. 查找正在运行任务的线程。这不是小事，但由于您可以访问所有堆栈帧，因此它当然是可行的。（如果在任务对象中放置唯一的id字段，则可以识别执行该字段的线程。）
3. Stop the thread asynchronously.

虽然我不认为停止的线程会严重干扰执行器（毕竟它们应该是故障安全的），但有一种替代解决方案不涉及停止线程

如果您的任务不修改系统其他部分中的任何内容（这是一个合理的假设，否则您不会试图将其击落），那么您可以使用JDI弹出不需要的堆栈帧并正常退出任务

public class StoppableTask implements Runnable {

private boolean stopped;
private Runnable targetTask;
private volatile Thread runner;
private String id;

public StoppableTask(TestTask targetTask) {
    this.targetTask = targetTask;
    this.id = UUID.randomUUID().toString();
}

@Override
public void run() {
    if( !stopped ) {
        runner = Thread.currentThread();
        targetTask.run();
    } else {
        System.out.println( "Task "+id+" stopped.");
    }
}

public Thread getRunner() {
    return runner;
}

public String getId() {
    return id;
}
}

这是包装所有其他runnable的runnable。它存储对正在执行的线程的引用（稍后将很重要）和一个id，以便我们可以通过JDI调用找到它

public class Main {

public static void main(String[] args) throws IOException, IllegalConnectorArgumentsException, InterruptedException, IncompatibleThreadStateException, InvalidTypeException, ClassNotLoadedException {
    //connect to the virtual machine
    VirtualMachineManager manager = Bootstrap.virtualMachineManager();
    VirtualMachine vm = null;
    for( AttachingConnector con : manager.attachingConnectors() ) {
        if( con instanceof SocketAttachingConnector ) {
            SocketAttachingConnector smac = (SocketAttachingConnector)con;
            Map<String,? extends Connector.Argument> arg = smac.defaultArguments();
            arg.get( "port" ).setValue( "8000");
            arg.get( "hostname" ).setValue( "localhost" );
            vm = smac.attach( arg );
        }
    }

    //start the test task
    ExecutorService service = Executors.newCachedThreadPool();
    StoppableTask task = new StoppableTask( new TestTask() );
    service.execute( task );
    Thread.sleep( 1000 );

    // iterate over all the threads
    for( ThreadReference thread : vm.allThreads() ) {
        //iterate over all the objects referencing the thread
        //could take a long time, limiting the number of referring
        //objects scanned is possible though, as not many objects will
        //reference our runner thread
        for( ObjectReference ob : thread.referringObjects( 0 ) ) {
            //this cast is safe, as no primitive values can reference a thread
            ReferenceType obType = (ReferenceType)ob.type();
            //if thread is referenced by a stoppable task
            if( obType.name().equals( StoppableTask.class.getName() ) ) {

                StringReference taskId = (StringReference)ob.getValue( obType.fieldByName( "id" ));

                if( task.getId().equals( taskId.value() ) ) {
                    //task with matching id found
                    System.out.println( "Task "+task.getId()+" found.");

                    //suspend thread
                    thread.suspend();

                    Iterator<StackFrame> it = thread.frames().iterator();
                    while( it.hasNext() ) {
                        StackFrame frame = it.next();
                        //find stack frame containing StoppableTask.run()
                        if( ob.equals( frame.thisObject() ) ) {
                            //pop all frames up to the frame below run()
                            thread.popFrames( it.next() );
                            //set stopped to true
                            ob.setValue( obType.fieldByName( "stopped") , vm.mirrorOf( true ) );
                            break;
                        }
                    }
                    //resume thread
                    thread.resume();

                }

            }
        }
    }

}
}

作为参考，我测试了“库”调用：

public class TestTask implements Runnable {

    @Override
    public void run() {
        long l = 0;
        while( true ) {
            l++;
            if( l % 1000000L == 0 )
                System.out.print( ".");
        }

    }
}

您可以通过使用命令行选项-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:8000,timeout=5000,suspend=n启动Main类来尝试。它有两个警告。首先，如果正在执行本机代码（thisObject帧的值为空），则必须等待它完成。其次，不会调用finally块，因此各种资源可能会泄漏