java - 多任务运行器算法性能不佳

Question

在我看来，使用我自己的以下 SupervisedExecutor 和 ExecutorSuperviser 实现我的性能很差，你认为这段代码中有什么地方效率低下？我想了解如何提高其效率。

ExecutorSuperviser 类:

public class ExecutorSuperviser {
private SupervisedExecutor[] threadPool;
private int poolSize = 0;
private LinkedList<Runnable> q;\\my own implementation of linkedlist
public ExecutorSuperviser(int nThreads) {
    threadPool=new SupervisedExecutor[poolSize=nThreads];
    q=new LinkedList<Runnable>();
    init();
}
public void execute(Runnable r) { 
    synchronized (q) {
        q.addToTail(r);
    }
    for (int i=0;i<poolSize;i++) 
            if (!threadPool[i].isBusy()) {
                if (!threadPool[i].isAlive()) threadPool[i].start();
                threadPool[i].interrupt();
                return;
            }


}
private void init() {
    for (int i=0;i<poolSize;i++) {
        threadPool[i]=new SupervisedExecutor(this);
    }

}
public Object getLock() {
    return q;
}
public Runnable getTask() {
    return q.removeHead();
}
public void terminate() {
    for (int i=0;i<poolSize;i++) 
        threadPool[i].terminate();
}
public void waitUntilFinished() {
    while (!isFinished()) {
        try {
            Thread.sleep(Thread.MAX_PRIORITY);
        } catch (InterruptedException e) {}
    }
}
private boolean isFinished() {
    for (int i=0;i<poolSize;i++) 
        if (threadPool[i].isBusy()) return false;
    return q.isEmpty();
}

}

SupervisedExecutor 类:

public class SupervisedExecutor extends Thread {
    private boolean   terminated = false;
    private Boolean busy = false;
    private ExecutorSuperviser boss;
    SupervisedExecutor (ExecutorSuperviser boss) {
        this.boss=boss;
    }
    public void run() {
        while (!terminated) {
            try {
                sleep(MAX_PRIORITY);
            } catch (InterruptedException e) {
                synchronized (busy) {
                    busy=true;
                }
                Runnable r;
                while (true) {
                    synchronized (boss.getLock()) {
                        r=boss.getTask();
                    }
                    if (r!=null) r.run();
                    else break;
                } 
                synchronized (busy) {
                    busy=false;
                }
            }
        }
    }

    public boolean isBusy() {
        boolean isBusy;
        synchronized (boss.getLock()) {
            isBusy=busy;
        }
        return isBusy;
    }
    public void terminate() {
        terminated=true;
    }

}

Answer 1

下面的解决方案有以下优点:

1. 作为 ThreadPoolExecutor 的子类，您不必重新实现 ThreadPoolExecutor 为您所做的一切，只是为了获得您所追求的 waitUntilFinished() 功能。

   <

2. 利用 ReentrantLock、Condition 和 await()/signal()您可以避免忙等待，这肯定会影响性能。

此实现通过利用 ThreadPoolExecutor 公开的 beforeExecute() 和 afterExecute() 方法来保持我们自己的 Activity 计数任务。我不使用 getActiveCount() 因为，根据 JavaDoc，它不能保证准确的答案(尽管在 ThreadPoolExecutor 的情况下它确实提供了准确的答案答案，我需要进一步研究才能确定)。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class WaitableThreadPoolExecutor extends ThreadPoolExecutor
{
    private Condition waitCondition;
    private ReentrantLock lock;
    private int taskCount = 0;

    public WaitableThreadPoolExecutor( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue )
    {
        super( corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue );

        lock = new ReentrantLock( );
        waitCondition = lock.newCondition( );
    }

    // if isEmpty() is true, then there is no need to block
    // otherwise, wait until waitCondition is signaled
    public void waitUntilFinished( )
    {
        lock.lock( );
        try
        {
            while ( !isEmpty( ) )
                waitCondition.await( );
        }
        catch ( InterruptedException e )
        {
            e.printStackTrace();
        }
        finally
        {
            lock.unlock( );
        }
    }

    // the ThreadPool is empty if our taskCount is 0 and the
    // work queue is empty (this may not be bullet-proof, for one
    // thing, I'm hesitant to use getActiveCount() because it
    // does not guarantee an exact answer
    protected boolean isEmpty( )
    {
        lock.lock( );
        try
        {
            return taskCount == 0 && getQueue( ).isEmpty( );
        }
        finally
        {
            lock.unlock( );
        }
    }

    // increment our task count before executing each task
    @Override
    protected void beforeExecute( Thread t, Runnable r )
    {
        super.beforeExecute( t, r );

        lock.lock( );
        try
        {
            taskCount += 1;
        }
        finally
        {
            lock.unlock( );
        }
    }

    // decrement our task count after executing each task
    // then, if the pool is empty, signal anyone waiting
    // on the waitCondition
    @Override
    protected void afterExecute( Runnable r, Throwable t )
    {
        super.afterExecute( r, t );

        lock.lock( );
        try
        {
            taskCount -= 1;

            if ( isEmpty( ) ) waitCondition.signalAll( );
        }
        finally
        {
            lock.unlock( );
        }
    }

    public static void main( String[] args )
    {
        WaitableThreadPoolExecutor pool = new WaitableThreadPoolExecutor( 2, 4, 5000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>( ) );

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

            pool.execute( new Runnable( )
            {
                @Override
                public void run( )
                {
                    try { Thread.sleep( (int) ( Math.random( ) * 5000 ) ); } catch ( InterruptedException e ) { }

                    System.out.println( threadId + " done." );
                }
            });
        }

        pool.waitUntilFinished( );

        System.out.println( "Done waiting." );
    }
}

我包含了一个简单的 main() 方法，您可以将其用作测试用例。它启动 10 个线程，这些线程在打印完成之前等待一段随机时间。然后主线程调用waitUntilFinished()。

结果看起来像(要点是 Done waiting. 总是最后打印:

1 done.
2 done.
0 done.
4 done.
3 done.
5 done.
7 done.
8 done.
6 done.
9 done.
Done waiting.