下面列出了java.util.concurrent.ThreadPoolExecutor#submit ( ) 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
private AtomicLong test_unit() throws InterruptedException {
final StatsItemSet statsItemSet = new StatsItemSet("topicTest", scheduler, null);
executor = new ThreadPoolExecutor(10, 20, 10, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(100), new ThreadFactoryImpl("testMultiThread"));
for (int i = 0; i < 10; i++) {
executor.submit(new Runnable() {
@Override
public void run() {
statsItemSet.addValue("topicTest", 2, 1);
}
});
}
while (true) {
if (executor.getCompletedTaskCount() == 10) {
break;
}
Thread.sleep(1000);
}
return statsItemSet.getStatsItem("topicTest").getValue();
}
public void write(final ILazyWriteableDataset out, final Dataset data, final SliceND slice) {
if (thread == null) {
try {
out.setSliceSync(null, data, slice);
} catch (DatasetException e) {
throw new RuntimeException(e);
}
} else {
ThreadPoolExecutor t = thread[nt++];
if (nt == thread.length) {
nt = 0;
}
WriteJob w = new WriteJob(out, data, slice);
task.add(w);
t.submit(w);
}
}
@SuppressWarnings("unused")
public static void main(String[] args) {
ThreadPoolExecutor pool = new ThreadPoolExecutor(1, Integer.MAX_VALUE, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());//Threads.newDaemonThreadFactory("htable"));
for (int i = 0; i < Integer.MAX_VALUE; ++i) {
System.out.println("index: " + i);
Future<?> future = pool.submit(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
@Test
public void test_threadPool() {
int total = 4;
ThreadPoolExecutor executor = new ThreadPoolExecutor(total, total, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
for (int i = 0; i < total; i++) {
executor.submit(new Task(i));
}
try {
executor.awaitTermination(1, TimeUnit.SECONDS);
} catch (Exception e) {
//
}
stopped = true;
executor.shutdown();
}
/**
* submit(runnable) throws RejectedExecutionException if saturated.
*/
public void testSaturatedSubmitRunnable() {
final CountDownLatch done = new CountDownLatch(1);
final ThreadPoolExecutor p =
new ThreadPoolExecutor(1, 1,
LONG_DELAY_MS, MILLISECONDS,
new ArrayBlockingQueue<Runnable>(1));
try (PoolCleaner cleaner = cleaner(p, done)) {
Runnable task = new CheckedRunnable() {
public void realRun() throws InterruptedException {
await(done);
}};
for (int i = 0; i < 2; ++i)
p.submit(task);
for (int i = 0; i < 2; ++i) {
try {
p.execute(task);
shouldThrow();
} catch (RejectedExecutionException success) {}
assertTrue(p.getTaskCount() <= 2);
}
}
}
/**
* submit(runnable) throws RejectedExecutionException if saturated.
*/
public void testSaturatedSubmitRunnable() {
final CountDownLatch done = new CountDownLatch(1);
final ThreadPoolExecutor p =
new ThreadPoolExecutor(1, 1,
LONG_DELAY_MS, MILLISECONDS,
new ArrayBlockingQueue<Runnable>(1));
try (PoolCleaner cleaner = cleaner(p, done)) {
Runnable task = new CheckedRunnable() {
public void realRun() throws InterruptedException {
await(done);
}};
for (int i = 0; i < 2; ++i)
p.submit(task);
for (int i = 0; i < 2; ++i) {
try {
p.execute(task);
shouldThrow();
} catch (RejectedExecutionException success) {}
assertTrue(p.getTaskCount() <= 2);
}
}
}
/**
* submit(callable) throws RejectedExecutionException if saturated.
*/
public void testSaturatedSubmitCallable() {
final CountDownLatch done = new CountDownLatch(1);
final ThreadPoolExecutor p =
new ThreadPoolExecutor(1, 1,
LONG_DELAY_MS, MILLISECONDS,
new ArrayBlockingQueue<Runnable>(1));
try (PoolCleaner cleaner = cleaner(p, done)) {
Runnable task = new CheckedRunnable() {
public void realRun() throws InterruptedException {
await(done);
}};
for (int i = 0; i < 2; ++i)
p.submit(Executors.callable(task));
for (int i = 0; i < 2; ++i) {
try {
p.execute(task);
shouldThrow();
} catch (RejectedExecutionException success) {}
assertTrue(p.getTaskCount() <= 2);
}
}
}
@Override
protected void serviceInit(Configuration conf) throws Exception
{
super.serviceInit(conf);
LOG.debug("Creating plugin dispatch queue with size {}", QUEUE_SIZE);
blockingQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);
RejectedExecutionHandler rejectionHandler = new RejectedExecutionHandler()
{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor)
{
try {
blockingQueue.remove();
executor.submit(r);
} catch (NoSuchElementException ex) {
// Ignore no-such element as queue may finish, while this handler is called.
}
}
};
executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
blockingQueue, new NameableThreadFactory("PluginExecutorThread", true), rejectionHandler);
}
@Override
public void run(final EnhancerModel enhancerModel) throws Exception {
if (started.compareAndSet(false, true)) {
oomExceptionCounter.init();
final JvmOomConfiguration jvmOomConfiguration = parse(enhancerModel);
executorService = new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "chaosblade-oom-thread");
}
});
executorService.submit(new Runnable() {
@Override
public void run() {
Long interval = calculateCostMemoryInterval(jvmOomConfiguration);
while (isStarted()) {
try {
innerRun(enhancerModel, jvmOomConfiguration);
if (interval > 0) {
TimeUnit.MILLISECONDS.sleep(interval);
}
} catch (Throwable throwable) {
handleThrowable(jvmOomConfiguration, throwable);
}
}
}
});
}
}
private Future<?> executeDiskStoreTask(DiskStoreTask r, ThreadPoolExecutor executor) {
try {
return executor.submit(r);
} catch (RejectedExecutionException ex) {
if (this.logger.fineEnabled()) {
this.logger.fine("Ignored compact schedule during shutdown", ex);
}
}
return null;
}
@Ignore
@Test
public void doTest() throws Exception {
ThreadPoolExecutor executor = new ThreadPoolExecutor(threadNum, threadNum, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(), new NamedThreadFactory("TestSessionNotify"));
for (int i = 0; i < idNum; i++) {
int finalI = i;
executor.submit(()->{
// post sync data request
DataChangeRequest request = new DataChangeRequest(DataInfo.toDataInfoId(
MockSyncDataHandler.dataId, DEFAULT_INSTANCE_ID, DEFAULT_GROUP), LOCAL_DATACENTER,
finalI);
boltChannelMap.forEach((connect,boltChannel)->{
CommonResponse commonResponse = null;
try {
dataSyncServer.sendCallback(dataSyncServer
.getChannel(new URL(boltChannel.getConnection().getLocalIP(), boltChannel.getConnection().getLocalPort())), request,
new NotifyCallback(boltChannel.getConnection(),request),3000);
//assertTrue(commonResponse.isSuccess());
} catch (Exception e) {
e.printStackTrace();
}
});
});
}
while (true){
TimeUnit.SECONDS.sleep(10);
}
}
public static void main(String[] args) {
//使用阿里巴巴推荐的创建线程池的方式
//通过ThreadPoolExecutor构造函数自定义参数创建
ThreadPoolExecutor executor = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
KEEP_ALIVE_TIME,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(QUEUE_CAPACITY),
new ThreadPoolExecutor.CallerRunsPolicy());
List<Future<String>> futureList = new ArrayList<>();
Callable<String> callable = new MyCallable();
for (int i = 0; i < 10; i++) {
//提交任务到线程池
Future<String> future = executor.submit(callable);
//将返回值 future 添加到 list，我们可以通过 future 获得 执行 Callable 得到的返回值
futureList.add(future);
}
for (Future<String> fut : futureList) {
try {
System.out.println(new Date() + "::" + fut.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
//关闭线程池
executor.shutdown();
}
public void start() {
threadPool = new ThreadPoolExecutor(2, Integer.MAX_VALUE, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName(String.format("locust4j-stats#%d#", threadNumber.getAndIncrement()));
return thread;
}
});
threadPool.submit(new StatsTimer(this));
threadPool.submit(this);
}
void test(String[] args) throws Throwable {
final int threadCount = 10;
final int timeoutMillis = 30;
BlockingQueue<Runnable> q = new ArrayBlockingQueue<>(2*threadCount);
ThreadPoolExecutor tpe
= new ThreadPoolExecutor(threadCount, threadCount,
timeoutMillis, TimeUnit.MILLISECONDS,
q, new IdentifiableThreadFactory());
equal(tpe.getCorePoolSize(), threadCount);
check(! tpe.allowsCoreThreadTimeOut());
tpe.allowCoreThreadTimeOut(true);
check(tpe.allowsCoreThreadTimeOut());
equal(countExecutorThreads(), 0);
long startTime = System.nanoTime();
for (int i = 0; i < threadCount; i++) {
tpe.submit(() -> {});
int count = countExecutorThreads();
if (millisElapsedSince(startTime) < timeoutMillis)
equal(count, i + 1);
}
while (countExecutorThreads() > 0 &&
millisElapsedSince(startTime) < LONG_DELAY_MS)
Thread.yield();
equal(countExecutorThreads(), 0);
check(millisElapsedSince(startTime) >= timeoutMillis);
tpe.shutdown();
check(tpe.allowsCoreThreadTimeOut());
check(tpe.awaitTermination(LONG_DELAY_MS, MILLISECONDS));
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
if (failed > 0) throw new Exception("Some tests failed");
}
private void processMappingQuery(final DatabaseSession databaseSession, QueryContext queryContext, GenerateGeometryResult generateGeometryResult, final StreamingSerializerPlugin ifcSerializerPlugin, final RenderEngineSettings settings,
final RenderEngineFilter renderEngineFilter, RenderEnginePool renderEnginePool, ThreadPoolExecutor executor, EClass eClass, Query query, QueryPart queryPart, boolean geometryReused, Map<Long, ProductDef> map, int nrObjects) throws QueryException, IOException {
JsonQueryObjectModelConverter jsonQueryObjectModelConverter = new JsonQueryObjectModelConverter(packageMetaData);
String queryNameSpace = packageMetaData.getSchema().name().toLowerCase() + "-stdlib";
if (eClass.getName().equals("IfcAnnotation")) {
// IfcAnnotation also has the field ContainedInStructure, but that is it's own field (looks like a hack on the IFC-spec side)
queryPart.addInclude(jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":IfcAnnotationContainedInStructure", true));
} else {
queryPart.addInclude(jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":ContainedInStructure", true));
}
if (packageMetaData.getSchema() == Schema.IFC4) {
queryPart.addInclude(jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":IsTypedBy", true));
}
queryPart.addInclude(jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":Decomposes", true));
queryPart.addInclude(jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":OwnerHistory", true));
Include representationInclude = jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":RepresentationSpecificMapping", true);
queryPart.addInclude(representationInclude);
Include objectPlacement = jsonQueryObjectModelConverter.getDefineFromFile(queryNameSpace + ":ObjectPlacement", true);
queryPart.addInclude(objectPlacement);
if (packageMetaData.getEClass("IfcElement").isSuperTypeOf(eClass)) {
Include openingsInclude = queryPart.createInclude();
openingsInclude.addType(packageMetaData.getEClass(eClass.getName()), false);
openingsInclude.addField("HasOpenings");
Include hasOpenings = openingsInclude.createInclude();
hasOpenings.addType(packageMetaData.getEClass("IfcRelVoidsElement"), false);
hasOpenings.addField("RelatedOpeningElement");
hasOpenings.addInclude(representationInclude);
hasOpenings.addInclude(objectPlacement);
// Include relatedOpeningElement = hasOpenings.createInclude();
// relatedOpeningElement.addType(packageMetaData.getEClass("IfcOpeningElement"), false);
// relatedOpeningElement.addField("HasFillings");
// Include hasFillings = relatedOpeningElement.createInclude();
// hasFillings.addType(packageMetaData.getEClass("IfcRelFillsElement"), false);
// hasFillings.addField("RelatedBuildingElement");
}
QueryObjectProvider queryObjectProvider = new QueryObjectProvider(databaseSession, bimServer, query, Collections.singleton(queryContext.getRoid()), packageMetaData);
ReportJob job = report.newJob(eClass.getName(), nrObjects);
GeometryRunner runner = new GeometryRunner(this, eClass, renderEnginePool, databaseSession, settings, queryObjectProvider, ifcSerializerPlugin, renderEngineFilter, generateGeometryResult, queryContext, geometryReused, map, job, reuseGeometry, geometryGenerationDebugger, query);
executor.submit(runner);
jobsTotal.incrementAndGet();
}
/**
* Execute la tache avec plusieurs Threads
*
* @param request
* @return
* @throws Exception
*/
private long execute(HttpGetRequestRunnable request, int numberOfRequests, int threads) throws Exception {
PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
httpClient =
HttpClientBuilder
.create()
.setConnectionManager(connectionManager)
.setMaxConnTotal(threads)
.setMaxConnPerRoute(threads)
.setDefaultRequestConfig(
RequestConfig.custom().setConnectTimeout(10000).setSocketTimeout(10000).build())
.build();
// Warm up
request.run();
BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(threads, threads, 5, TimeUnit.SECONDS, queue);
long start = System.currentTimeMillis();
threadPool.prestartAllCoreThreads();
for (int i = 0; i < numberOfRequests; i++) {
threadPool.submit(request);
}
threadPool.shutdown();
// wait maximum 20 s
threadPool.awaitTermination(200, TimeUnit.SECONDS);
connectionManager.shutdown();
if (request.exception != null) {
throw new AssertionFailedError("Exception for request " + request.url + " after " + request.count
+ " requests", request.exception);
}
if (threadPool.getCompletedTaskCount() < threadPool.getTaskCount()) {
// All task were not executed
String msg =
request.url + " : Only " + threadPool.getCompletedTaskCount() + "/" + threadPool.getTaskCount()
+ " have been renderered " + " => Maybe a performance issue";
threadPool.shutdownNow();
fail(msg);
}
long end = System.currentTimeMillis();
long execTime = end - start;
LOG.debug("Executed request " + request.url + " " + numberOfRequests + " times with " + threads
+ " threads in " + execTime + "ms");
return execTime;
}
public void testResubmittingRunnable()
{
final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
Runnable r = new Runnable()
{
public void run()
{
// do nothing... used to delay the executor;
sleep(100);
}
};
final int[]ran = new int[1];
final boolean[]bool = new boolean[1];
Runnable r2 = new Runnable()
{
public void run()
{
ran[0]++;
if (ran[0] < 5)
{
if (executor.isShutdown())
{
bool[0] = true;
}
executor.getQueue().add(this);
}
}
};
for(int i=0;i<10;i++)
{
executor.submit(r);
}
executor.submit(r2);
executor.shutdown();
try
{
executor.awaitTermination(100, TimeUnit.SECONDS);
}
catch (InterruptedException e)
{
fail();
}
assertTrue(bool[0]);
assertEquals(5, ran[0]);
}
@Test (timeout = 120000)
public void testPriorityBasedExecutor() throws Exception {
List<Priority> priorityList = Collections.synchronizedList(new ArrayList<Priority>());
ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat("priority-exec-thread-%d").build();
ThreadPoolExecutor threadPoolExecutor = new PriorityBasedThreadPoolExecutor(2, 2, 0L,
TimeUnit.MILLISECONDS, factory);
Future<?> future1 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.LOW), Priority.LOW));
Future<?> future2 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.LOW), Priority.LOW));
Future<?> future3 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.LOW), Priority.LOW));
Future<?> future4 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.LOW), Priority.LOW));
Future<?> future5 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.NORMAL), Priority.NORMAL));
Future<?> future6 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.NORMAL), Priority.NORMAL));
Future<?> future7 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.HIGH), Priority.HIGH));
Future<?> future8 = threadPoolExecutor.submit(
RunnableWithPriority.get(new DummyRunnable(priorityList, Priority.HIGH), Priority.HIGH));
List<Priority> expectedPriorityList = Lists.newArrayList(Priority.LOW, Priority.LOW);
while (priorityList.size() < 2) {
Thread.sleep(100);
}
Assert.assertEquals(expectedPriorityList, priorityList);
future1.cancel(true);
future2.cancel(true);
expectedPriorityList.addAll(Lists.newArrayList(Priority.HIGH, Priority.HIGH));
while (priorityList.size() < 4) {
Thread.sleep(100);
}
Assert.assertEquals(expectedPriorityList, priorityList);
future7.cancel(true);
future8.cancel(true);
expectedPriorityList.addAll(Lists.newArrayList(Priority.NORMAL, Priority.NORMAL));
while (priorityList.size() < 6) {
Thread.sleep(100);
}
Assert.assertEquals(expectedPriorityList, priorityList);
future5.cancel(true);
future6.cancel(true);
expectedPriorityList.addAll(Lists.newArrayList(Priority.LOW, Priority.LOW));
while (priorityList.size() < 8) {
Thread.sleep(100);
}
Assert.assertEquals(expectedPriorityList, priorityList);
future3.cancel(true);
future4.cancel(true);
threadPoolExecutor.shutdownNow();
}
public boolean getPage(String url, String outputDirPath, String indexFilename) {
this.indexFileName = indexFilename;
File outputDir = new File(outputDirPath);
if (!outputDir.exists() && outputDir.mkdirs() == false) {
eventCallback.onFatalError(new IOException("File " + outputDirPath + "could not be created"), url);
return false;
}
//download main html and parse -- isExtra parameter should be false
boolean success = downloadHtmlAndParseLinks(url, outputDirPath, false);
if (isCancelled || !success) {
return false;
}
//download and parse html frames - use iterator because our list may be modified as frames can contain other frames
for (Iterator<String> i = framesToGrab.iterator(); i.hasNext();) {
downloadHtmlAndParseLinks(i.next(), outputDirPath, true);
if (isCancelled) return true;
}
//download and parse css files
for (Iterator<String> i = cssToGrab.iterator(); i.hasNext();) {
if (isCancelled) return true;
downloadCssAndParse(i.next(), outputDirPath);
}
ThreadPoolExecutor pool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 60, TimeUnit.SECONDS, new BlockingDownloadTaskQueue<Runnable>());
for (Iterator<String> i = filesToGrab.iterator(); i.hasNext();) {
if (isCancelled) {
eventCallback.onProgressMessage("Cancelling...");
shutdownExecutor(pool, 10, TimeUnit.SECONDS);
return success;
}
String urlToDownload = i.next();
eventCallback.onProgressMessage("Saving file: " + getFileName(urlToDownload));
eventCallback.onProgressChanged(filesToGrab.indexOf(urlToDownload), filesToGrab.size(), false);
pool.submit(new DownloadTask(urlToDownload, outputDir));
}
pool.submit(new DownloadTask(pageIconUrl, outputDir, "saveForOffline_icon.png"));
eventCallback.onProgressMessage("Finishing file downloads...");
shutdownExecutor(pool, 60, TimeUnit.SECONDS);
return success;
}
@Test
public void testFixedTimingOutThreadPool() throws Exception {
int numMaxThreads = 5;
long timeoutMillis = 2;
ThreadPoolExecutor es = JavaUtils.newFixedTimingOutThreadPool(
numMaxThreads,
timeoutMillis,
"test"
);
Thread.sleep(timeoutMillis + 100);
Assertions.assertTrue(es.getPoolSize() == 0);
final CountDownLatch block = new CountDownLatch(1);
final CountDownLatch cdl = new CountDownLatch(numMaxThreads);
for (int i = 0; i < numMaxThreads; i++) {
es.submit(
new Runnable() {
@Override
public void run() {
try {
cdl.countDown();
cdl.await();
block.await();
} catch (Throwable t) {
throw new RuntimeException(t);
}
}
}
);
}
cdl.await();
// all threads are running:
Assertions.assertTrue(es.getPoolSize() == numMaxThreads);
block.countDown();
Thread.sleep(timeoutMillis + 100);
Assertions.assertTrue(es.getMaximumPoolSize() == numMaxThreads);
Assertions.assertTrue(es.getPoolSize() == 0);
es.shutdown();
es.awaitTermination(timeoutMillis + 1, TimeUnit.MILLISECONDS);
}