晋中品牌网站建设建设,网络营销课程自学,网站正在建设中 免费,唐山网站建设选汉狮自定义java线程池ThreadPoolExecutor是Java并发api添加的一项功能#xff0c;可以有效地维护和重用线程#xff0c;因此我们的程序不必担心创建和销毁线程#xff0c;也不必关注核心功能。 我创建了一个自定义线程池执行程序#xff0c;以更好地了解线程池执行程序的工作方… 自定义java线程池 ThreadPoolExecutor是Java并发api添加的一项功能可以有效地维护和重用线程因此我们的程序不必担心创建和销毁线程也不必关注核心功能。 我创建了一个自定义线程池执行程序以更好地了解线程池执行程序的工作方式。 功能性 它维护一个固定的线程池即使没有任务提交也创建线程并启动线程而ThreadPoolExecutor根据需要创建线程即每当将可运行对象提交给池且线程数小于核心池大小时。 在ThreadPoolExecutor中我们提供了一个等待队列当所有线程忙于运行现有任务时新的可运行任务将在该队列中等待。 队列填满后将创建最大线程池大小的新线程。 在MyThreadPool中我将可运行对象存储在链接列表中因此每个任务都将在列表中等待并且不受限制因此在此不使用maxPoolSize。 在ThreadPoolExecutor中我们使用Future Objects从任务中获取结果如果结果不可用则future.get方法将阻塞或者使用CompletionService。 在MyThreadPoolExecutor中我创建了一个名为ResultListener的简单接口用户必须提供对此的实现如他希望如何处理输出。 每个任务完成后ResultListener将获得带有任务输出的回调或者在发生任何异常的情况下将调用error方法。 调用shutdown方法时MyThreadPoolExecutor将停止接受新任务并完成剩余任务。 与ThreadPoolExecutor相比我提供了非常基本的功能我使用了简单的线程机制如waitnotifynotifyAll和join。 在性能方面它类似于ThreadPoolExecutor在某些情况下好一些。 如果您发现任何有趣的结果或改进方法请告诉我。 package com.util;import java.util.concurrent.Callable;/*** Run submitted task of {link MyThreadPool} After running the task , It calls* on {link ResultListener}object with {link Output}which contains returned* result of {link Callable}task. Waits if the pool is empty.* * author abhishek* * param */import java.util.concurrent.Callable;
/**
* Run submitted task of {link MyThreadPool} After running the task , It calls
* on {link ResultListener}object with {link Output}which contains returned
* result of {link Callable}task. Waits if the pool is empty.
*
* author abhishek
*
* param V
*/
public class MyThreadV extends Thread {/*** MyThreadPool object, from which the task to be run*/private MyThreadPoolV pool;private boolean active true;public boolean isActive() {return active;}public void setPool(MyThreadPoolV p) {pool p;}/*** Checks if there are any unfinished tasks left. if there are , then runs* the task and call back with output on resultListner Waits if there are no* tasks available to run If shutDown is called on MyThreadPool, all waiting* threads will exit and all running threads will exit after finishing the* task*/public void run() {ResultListenerV result pool.getResultListener();CallableV task;while (true){task pool.removeFromQueue();if (task ! null){try{V output task.call();result.finish(output);} catch (Exception e){result.error(e);}} else{if (!isActive())break;else{synchronized (pool.getWaitLock()){try{pool.getWaitLock().wait();} catch (InterruptedException e){// TODO Auto-generated catch blocke.printStackTrace();}}}}}}void shutdown() {active false;}
}package com.util;
import java.util.LinkedList;
import java.util.concurrent.Callable;
/**
* This class is used to execute submitted {link Callable} tasks. this class
* creates and manages fixed number of threads User will provide a
* {link ResultListener}object in order to get the Result of submitted task
*
* author abhishek
*
*
*/
public class MyThreadPoolV {private Object waitLock new Object();public Object getWaitLock() {return waitLock;}/*** list of threads for completing submitted tasks*/private final LinkedListMyThreadV threads;/*** submitted task will be kept in this list untill they run by one of* threads in pool*/private final LinkedListCallableV tasks;/*** shutDown flag to shut Down service*/private volatile boolean shutDown;/*** ResultListener to get back the result of submitted tasks*/private ResultListenerV resultListener;/*** initializes the threadPool by starting the threads threads will wait till* tasks are not submitted** param size* Number of threads to be created and maintained in pool* param myResultListener* ResultListener to get back result*/public MyThreadPool(int size, ResultListenerV myResultListener) {tasks new LinkedListCallableV();threads new LinkedListMyThreadV();shutDown false;resultListener myResultListener;for (int i 0; i size; i) {MyThreadV myThread new MyThreadV();myThread.setPool(this);threads.add(myThread);myThread.start();}}public ResultListenerV getResultListener() {return resultListener;}public void setResultListener(ResultListenerV resultListener) {this.resultListener resultListener;}public boolean isShutDown() {return shutDown;}public int getThreadPoolSize() {return threads.size();}public synchronized CallableV removeFromQueue() {return tasks.poll();}public synchronized void addToTasks(CallableV callable) {tasks.add(callable);}/*** submits the task to threadPool. will not accept any new task if shutDown* is called Adds the task to the list and notify any waiting threads** param callable*/public void submit(CallableV callable) {if (!shutDown) {addToTasks(callable);synchronized (this.waitLock) {waitLock.notify();}} else {System.out.println(task is rejected.. Pool shutDown executed);}}/*** Initiates a shutdown in which previously submitted tasks are executed,* but no new tasks will be accepted. Waits if there are unfinished tasks* remaining**/public void stop() {for (MyThreadV mythread : threads) {mythread.shutdown();}synchronized (this.waitLock) {waitLock.notifyAll();}for (MyThreadV mythread : threads) {try {mythread.join();} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}
}package com.util;/*** This interface imposes finish method * which is used to get the {link Output} object * of finished task* author abhishek** param */public interface ResultListener {public void finish(T obj);public void error(Exception ex);} 您可以根据需要实现此类并返回并处理任务返回的结果。 package com.util;public class DefaultResultListener implements ResultListener{Overridepublic void finish(Object obj) {}Overridepublic void error(Exception ex) {ex.printStackTrace();}} 例如此类将添加task返回的数字。 package com.util;import java.util.concurrent.atomic.AtomicInteger;/*** ResultListener class to keep track of total matched count* author abhishek* * param */
public class MatchedCountResultListenerimplements ResultListener{/*** matchedCount to keep track of the number of matches returned by submitted* task*/AtomicInteger matchedCount new AtomicInteger();/*** this method is called by ThreadPool to give back the result of callable* task. if the task completed successfully then increment the matchedCount by* result count*/Overridepublic void finish(V obj) {//System.out.println(count is obj);matchedCount.addAndGet((Integer)obj);}/*** print exception thrown in running the task*/Overridepublic void error(Exception ex) {ex.printStackTrace();}/*** returns the final matched count of all the finished tasks* * return*/public int getFinalCount() {return matchedCount.get();}
} 这是一个测试类使用CompletionService和MyThreadPoolExecutor对循环运行简单 package test;import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;import com.util.DefaultResultListener;
import com.util.MyThreadPool;public class TestClass {public static void main(String[] args) throws InterruptedException {CompletionServicethreadService;ExecutorService service Executors.newFixedThreadPool(2);threadService new ExecutorCompletionService(service);long b System.currentTimeMillis();for(int i 0;i50000;i){threadService.submit(new MyRunable (i));}service.shutdown();System.out.println(time taken by Completion Service (System.currentTimeMillis()-b));DefaultResultListener result new DefaultResultListener();MyThreadPoolnewPool new MyThreadPool(2,result);long a System.currentTimeMillis();int cc 0;for(int i 0;i50000;i){cc cci;}System.out.println(time taken without any pool (System.currentTimeMillis()-a));a System.currentTimeMillis();for(int i 0;i5000;i){newPool.submit(new MyRunable (i));}newPool.stop();System.out.println(time taken by myThreadPool (System.currentTimeMillis()-a));}}class MyRunable implements Callable{int index -1;public MyRunable(int index){this.index index;}Overridepublic Integer call() throws Exception {return index;}} 参考 我的JCG合作伙伴 Abhishek Somani在JavaJ2EE和Server博客上的Java 自定义线程池执行程序 。 翻译自: https://www.javacodegeeks.com/2013/03/my-custom-thread-pool-executor-in-java.html自定义java线程池
