|
public class Threads {
static ExecutorService executorService = Executors . newFixedThreadPool ( 1 ) ;
@SuppressWarnings ( “rawtypes” )
public static void main (String[] args ) throws InterruptedException , ExecutionException
{
SubThread thread = new SubThread () ;
// thread.start();
Future future = executorService . submit (thread) ;
mainThreadOtherWork () ;
System . out . println ( “now
waiting sub thread done.” ) ;
future . get () ;
// try {
// thread.join();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
System . out . println ( “now
all done.” ) ;
executorService . shutdown () ;
}
private static void mainThreadOtherWork ()
{
System . out . println ( “main
thread work start” ) ;
try {
Thread . sleep ( 3000L ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
System . out . println ( “main
thread work done.” ) ;
}
public static class SubThread extends Thread{
@Override
public void run ()
{
working () ;
}
private void working ()
{
System . out . println ( “sub
thread start working.” ) ;
busy () ;
System . out . println ( “sub
thread stop working.” ) ;
}
private void busy ()
{
try {
sleep ( 5000L ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
}
}
}
這 里, ThreadPoolExecutor 是實(shí)現(xiàn)了 ExecutorService的方法, sumbit的過程就是把一個(gè)Runnable接口對(duì)象包裝成一個(gè) Callable接口對(duì)象, 然后放到 workQueue里等待調(diào)度執(zhí)行. 當(dāng)然, 執(zhí)行的啟動(dòng)也是調(diào)用了thread的start來做到的, 只不過這里被包裝掉了. 另外, 這里的thread是會(huì)被重復(fù)利用的, 所以這里要退出主線程, 需要執(zhí)行以下shutdown方法以示退出使用線程池.
扯遠(yuǎn)了.
這 種方法是得益于Callable接口和Future模式, 調(diào)用future接口的get方法, 會(huì)同步等待該future執(zhí)行結(jié)束, 然后獲取到結(jié)果. Callbale接口的接口方法是 V call(); 是可以有返回結(jié)果的, 而Runnable的 void run(), 是沒有返回結(jié)果的. 所以, 這里即使被包裝成Callbale接口, future.get返回的結(jié)果也是null的.如果需要得到返回結(jié)果, 建議使用Callable接口.
通過隊(duì)列來控制線程的進(jìn)度, 是很好的一個(gè)理念. 我們完全可以自己搞個(gè)隊(duì)列, 自己控制. 這樣也可以實(shí)現(xiàn). 不信看代碼:
public class Threads {
// static ExecutorService executorService = Executors.newFixedThreadPool(1);
static final BlockingQueue < Integer > queue = new ArrayBlockingQueue < Integer > ( 1 ) ;
public static void main (String[] args ) throws InterruptedException , ExecutionException
{
SubThread thread = new SubThread ( queue ) ;
thread . start () ;
// Future future = executorService.submit(thread);
mainThreadOtherWork () ;
System . out . println ( “now
waiting sub thread done.” ) ;
// future.get();
queue . take () ;
// try {
// thread.join();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
System . out . println ( “now
all done.” ) ;
// executorService.shutdown();
}
private static void mainThreadOtherWork () {
System . out . println ( “main
thread work start” ) ;
try {
Thread . sleep ( 3000L ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
System . out . println ( “main
thread work done.” ) ;
}
public static class SubThread extends Thread{
private BlockingQueue < Integer > queue ;
/**
* @param queue
*/
public SubThread ( BlockingQueue < Integer > queue )
{
this . queue = queue ;
}
@Override
public void run () {
try {
working () ;
} finally {
try {
queue . put ( 1 ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
}
}
private void working () {
System . out . println ( “sub
thread start working.” ) ;
busy () ;
System . out . println ( “sub
thread stop working.” ) ;
}
private void busy () {
try {
sleep ( 5000L ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
}
}
}
這 里是得益于我們用了一個(gè)阻塞隊(duì)列, 他的put操作和take操作都會(huì)阻塞(同步), 在滿足條件的情況下.當(dāng)我們調(diào)用take()方法是, 由于子線程還沒結(jié)束, 隊(duì)列是空的, 所以這里的take操作會(huì)阻塞, 直到子線程結(jié)束的時(shí)候, 往隊(duì)列里面put了個(gè)元素, 表明自己結(jié)束了. 這時(shí)候主線程的take()就會(huì)返回他拿到的數(shù)據(jù). 當(dāng)然, 他拿到什么我們是不必去關(guān)心的.
以上幾種情況都是針對(duì)子線程只有1個(gè)的時(shí)候. 當(dāng)子線程有多個(gè)的時(shí)候, 情況就不妙了.
第一種方法, 你要調(diào)用很多個(gè)線程的join, 特別是當(dāng)你的線程不是for循環(huán)創(chuàng)建的, 而是一個(gè)一個(gè)創(chuàng)建的時(shí)候.
第二種方法, 要調(diào)用很多的future的get方法, 同第一種方法.
第三種方法, 比較方便一些, 只需要每個(gè)線程都在queue里面 put一個(gè)元素就好了.但是, 第三種方法, 這個(gè)隊(duì)列里的對(duì)象, 對(duì)我們是毫無用處, 我們?yōu)榱耸褂藐?duì)列, 而要不明不白浪費(fèi)一些內(nèi)存, 那有沒有更好的辦法呢?
有的, concurrency包里面提供了好多有用的東東, 其中, CountDownLanch就是我們要用的.
CountDownLanch 是一個(gè)倒數(shù)計(jì)數(shù)器, 給一個(gè)初始值(>=0), 然后沒countDown一次就會(huì)減1, 這很符合等待多個(gè)子線程結(jié)束的產(chǎn)景: 一個(gè)線程結(jié)束的時(shí)候, countDown一次, 直到所有都countDown了 , 那么所有子線程就都結(jié)束了.
先看看CountDownLanch有哪些方法:
 await: 會(huì)阻塞等待計(jì)數(shù)器減少到0位置. 帶參數(shù)的await是多了等待時(shí)間.
countDown: 將當(dāng)前的技術(shù)減1
getCount(): 返回當(dāng)前的計(jì)數(shù)
顯而易見, 我們只需要在子線程執(zhí)行之前, 賦予初始化countDownLanch, 并賦予線程數(shù)量為初始值.
每個(gè)線程執(zhí)行完畢的時(shí)候, 就countDown一下.主線程只需要調(diào)用await方法, 可以等待所有子線程執(zhí)行結(jié)束, 看代碼:
public class Threads {
// static ExecutorService executorService = Executors.newFixedThreadPool(1);
static final BlockingQueue < Integer > queue = new ArrayBlockingQueue < Integer > ( 1 ) ;
public static void main (String[] args ) throws InterruptedException , ExecutionException
{
int threads = 5 ;
CountDownLatch countDownLatch = new CountDownLatch (threads) ;
for ( int i = 0 ; i < threads ; i ++ ){
SubThread thread = new SubThread ( 2000 * (i + 1 ) , countDownLatch) ;
thread . start () ;
}
// Future future = executorService.submit(thread);
mainThreadOtherWork () ;
System . out . println ( “now
waiting sub thread done.” ) ;
// future.get();
// queue.take();
countDownLatch . await () ;
// try {
// thread.join();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
System . out . println ( “now
all done.” ) ;
// executorService.shutdown();
}
private static void mainThreadOtherWork ()
{
System . out . println ( “main
thread work start” ) ;
try {
Thread . sleep ( 3000L ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
System . out . println ( “main
thread work done.” ) ;
}
public static class SubThread extends Thread{
// private BlockingQueue<Integer> queue;
private CountDownLatch countDownLatch ;
private long work ;
/**
* @param queue
*/
// public SubThread(BlockingQueue<Integer> queue) {
// this.queue = queue;
// this.work = 5000L;
// }
public SubThread ( long work , CountDownLatch countDownLatch )
{
// this.queue = queue;
this . countDownLatch = countDownLatch ;
this . work = work ;
}
@Override
public void run () {
try {
working () ;
} finally {
// try {
// queue.put(1);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
countDownLatch . countDown () ;
}
}
private void working () {
System . out . println ( getName () + ”
sub thread start working.” ) ;
busy () ;
System . out . println ( getName () + ”
sub thread stop working.” ) ;
}
private void busy () {
try {
sleep ( work ) ;
} catch (InterruptedException e) {
e . printStackTrace () ;
}
}
}
}
|
|
|
