Java:类完全在第二个线程/IllegalMonitorStateException 中运行

Question

当你想让某个任务由另一个线程执行时，你可以扩展Thread或者实现Runnable。

我尝试创建一个完全在第二个线程中运行类的类。

这意味着您可以调用立即返回并由第二个线程执行的anyMethod()。

这是我的尝试:

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Extend this class to run method calls asynchronously in the second thread implemented by this class.
 * Create method(type1 param1, type2 param2, ...) and let it call this.enqueueVoidCall("method", param1, param2, ...)
 * 
 * The thread executing the run-method will automatically call methodAsync with the specified parameters.
 * To obtain the return-value, pass an implementation of AsyncCallback to this.enqueueCall().
 * AsyncCallback.returnValue() will automatically be called upon completion of the methodAsync.
 *  
 */
public class ThreadedClass extends Thread {
    private static Object test;

    private Queue<String> queue_methods = new ConcurrentLinkedQueue<String>();
    private Queue<Object[]> queue_params = new ConcurrentLinkedQueue<Object[]>();
    private Queue<AsyncCallback<? extends Object>> queue_callback = new ConcurrentLinkedQueue<AsyncCallback<? extends Object>>();

    private volatile boolean shutdown = false;

/**
 *  The run method required by Runnable. It manages the asynchronous calls placed to this class.
 */
@Override
public final void run() {
    test = new Object();
    while (!shutdown) {
        if (!this.queue_methods.isEmpty()) {
            String crtMethod = queue_methods.poll();
            Object[] crtParamArr = queue_params.poll();
            String methodName = crtMethod + "Async";

            Method method;
            try {
                method = this.getClass().getMethod(methodName);
                try {
                    Object retVal = method.invoke(this, crtParamArr);
                    AsyncCallback<? extends Object> crtCallback = queue_callback.poll();
                    crtCallback.returnValue(retVal);
                } catch (Exception ex) {}
               } catch (SecurityException ex) {
               } catch (NoSuchMethodException ex) {}
        } else {
            try {
                synchronized(test ) {
                    test.wait();
                }
            } catch (InterruptedException ex) {
                System.out.println("READY");
            } catch (Exception ex) {
                System.out.println("READY, but " + ex.getMessage());
            }
        }
    }
}

/**
 * Asynchronously adds a method-call to the scheduler, specified by methodName with passed parameters 
 * @param methodName The name of the currently called method. methodName + "Async" is being called
 * @param parameters Parameters you may want to pass to the method
 */
protected final void enqueueVoidCall(String methodName, Object... parameters) {
    List<Object> tmpParam = new ArrayList<Object>();
    for (Object crt : parameters) {
        tmpParam.add(crt);
    }
    queue_methods.add(methodName);
    queue_params.add(parameters);
    queue_callback.add(null);
    test.notifyAll();
}

/**
 * Asynchronously adds a method-call to the scheduler, specified by methodName with passed parameters 
 * @param methodName The name of the currently called method. methodName + "Async" is being called
 * @param callBack An instance of AsyncCallback whose returnValue-method is called upon completion of the task.
 * @param parameters Parameters you may want to pass to the method
 */
protected final void enqueueCall(String methodName, AsyncCallback<? extends Object> callBack, Object... parameters) {
    List<Object> tmpParam = new ArrayList<Object>();
    for (Object crt : parameters) {
        tmpParam.add(crt);
    }
    queue_methods.add(methodName);
    queue_params.add(parameters);
    queue_callback.add(callBack);
    test.notifyAll();
}

/**
 * Complete the currently running task, optionally return values and eventually shut down. The instance of this object becomes unusable after this call. 
 */
public void shutdown() {
    shutdown=true;
}

}

现在我有两个类来测试:

public class MySecondTask extends ThreadedClass {
public void test1() {
    this.enqueueVoidCall("test1", null);
}

public void test1Async() {
    System.out.println("Start");
    try {
        // do big job here
    } catch (Exception ex) { }
    System.out.println("Done");
}
}

以及启动这些东西的主要方法:

public class TestingClass {
public static void main(String[] args) {
    MySecondTask test = new MySecondTask();
    test.start();
    System.out.println("1. Thread [1]");
    // CORRECTION, SHOULD BE:
    test.test1();
    // INSTEAD OF:
    // test.test1Async();
    for(int q=0; q<=100000; q++) {
        System.out.println("1:"+ new Date().getTime()+":"+ q);
        if ((q % 1000) == 0) {
            System.out.flush();
        }
    }
    System.err.println("1. Thread [2]");
}

}

不知何故，第二个线程的输出总是首先(全部)出现，然后其余的输出在控制台上。如果线程同时运行(这是预期的结果)，控制台输出应该是混合的？!

任何想法以及改进我的编码风格的评论都会受到赞赏。

<小时/>

编辑:

所提到的问题已经完全解决了。

现在我在调用 ThreadedClass.notifyAll() 的线路上收到 IllegalMonitorStateException。

也许我的锁锁错了。

但是a)为什么需要在wait()周围使用synchronized()以及如何调用notifyAll()来解锁wait()？

<小时/>

提前致谢并致以最诚挚的问候

p.s.:你们在堆栈溢出方面都做得很好。你已经在不知不觉中帮助了我很多次，谢谢你。坚持下去!

Answer 1

This means that you can call anyMethod() which returns immediately and which is executed by the second thread.

这听起来很像使用可调用对象、futures 和执行器:

• http://java.sun.com/docs/books/tutorial/essential/concurrency/pools.html
• http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/Callable.html
• http://java.sun.com/javase/6/docs/api/java/util/concurrent/ExecutorService.html
• http://java.sun.com/javase/6/docs/api/java/util/concurrent/Future.html

不想透露给你，但你可能真的想研究一下这个东西..

编辑以解决下面的评论

只需使对象中的方法如下所示:

private ExecutorService executorService = Executors.newCachedThreadPool();

public Future<SomeObject> yourMethodName(String anyArguments) {
    return executorService.submit(
        new Callable<SomeObject>() {
            public SomeObject call() {
                SomeObject obj = new SomeObject();
                /* Your Code Here*/;
                return obj;
            }
        }
    );
}