import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
/**  
	* thread pool of consumer
	*/
	@Bean(name="defaultThreadPoolProfile")
	ThreadPoolProfile threadPoolProfile(){
		ThreadPoolProfile defaultThreadPoolProfile = new ThreadPoolProfile();
		defaultThreadPoolProfile.setDefaultProfile(true);
		defaultThreadPoolProfile.setId("defaultThreadPoolProfile");
		defaultThreadPoolProfile.setPoolSize(threadPoolSize);
		defaultThreadPoolProfile.setMaxPoolSize(threadMaxPoolSize);
		defaultThreadPoolProfile.setMaxQueueSize(threadMaxQueueSize); // 队列最大程度1000万
		defaultThreadPoolProfile.setTimeUnit(TimeUnit.SECONDS);
		defaultThreadPoolProfile.setKeepAliveTime(60 * 5L);
		defaultThreadPoolProfile.setRejectedPolicy(ThreadPoolRejectedPolicy.CallerRuns);
//		camelContext().getExecutorServiceManager().registerThreadPoolProfile(defaultThreadPoolProfile);
//		setDefaultThreadPoolProfile(defaultThreadPoolProfile);
		return defaultThreadPoolProfile;
	}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public ExecutorService newThreadPool(Object source, String name, int corePoolSize, int maxPoolSize,
                                     long keepAliveTime, TimeUnit timeUnit, int maxQueueSize,
                                     RejectedExecutionHandler rejectedExecutionHandler, boolean daemon) {
    // use a profile with the settings
    ThreadPoolProfile profile = new ThreadPoolProfile();
    profile.setPoolSize(corePoolSize);
    profile.setMaxPoolSize(maxPoolSize);
    profile.setMaxQueueSize(maxQueueSize);
    profile.setKeepAliveTime(keepAliveTime);
    profile.setTimeUnit(timeUnit);

    // must cast to ThreadPoolExecutor to be able to set the rejected execution handler
    ThreadPoolExecutor answer = (ThreadPoolExecutor) camelContext.getExecutorServiceManager().newThreadPool(source, name, profile);
    answer.setRejectedExecutionHandler(rejectedExecutionHandler);
    return answer;
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
@Override
public ExecutorService newThreadPool(Object source, String name, ThreadPoolProfile profile) {
    String sanitizedName = URISupport.sanitizeUri(name);
    ObjectHelper.notNull(profile, "ThreadPoolProfile");

    ThreadPoolProfile defaultProfile = getDefaultThreadPoolProfile();
    profile.addDefaults(defaultProfile);

    ThreadFactory threadFactory = createThreadFactory(sanitizedName, true);
    ExecutorService executorService = threadPoolFactory.newThreadPool(profile, threadFactory);
    onThreadPoolCreated(executorService, source, profile.getId());
    if (LOG.isDebugEnabled()) {
        LOG.debug("Created new ThreadPool for source: {} with name: {}. -> {}", source, sanitizedName, executorService);
    }

    return executorService;
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
@Override
public ScheduledExecutorService newScheduledThreadPool(ThreadPoolProfile profile, ThreadFactory threadFactory) {
    RejectedExecutionHandler rejectedExecutionHandler = profile.getRejectedExecutionHandler();
    if (rejectedExecutionHandler == null) {
        rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
    }

    ScheduledThreadPoolExecutor answer = new RejectableScheduledThreadPoolExecutor(profile.getPoolSize(), threadFactory, rejectedExecutionHandler);
    answer.setRemoveOnCancelPolicy(true);

    // need to wrap the thread pool in a sized to guard against the problem that the
    // JDK created thread pool has an unbounded queue (see class javadoc), which mean
    // we could potentially keep adding tasks, and run out of memory.
    if (profile.getMaxPoolSize() > 0) {
        return new SizedScheduledExecutorService(answer, profile.getMaxQueueSize());
    } else {
        return answer;
    }
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
protected synchronized ScheduledExecutorService getExecutorService(CamelContext camelContext) {
    if (executorService == null || executorService.isShutdown()) {
        // camel context will shutdown the executor when it shutdown so no need to shut it down when stopping
        if (executorServiceRef != null) {
            executorService = camelContext.getRegistry().lookupByNameAndType(executorServiceRef, ScheduledExecutorService.class);
            if (executorService == null) {
                ExecutorServiceManager manager = camelContext.getExecutorServiceManager();
                ThreadPoolProfile profile = manager.getThreadPoolProfile(executorServiceRef);
                executorService = manager.newScheduledThreadPool(this, executorServiceRef, profile);
            }
            if (executorService == null) {
                throw new IllegalArgumentException("ExecutorServiceRef " + executorServiceRef + " not found in registry.");
            }
        } else {
            // no explicit configured thread pool, so leave it up to the error handler to decide if it need
            // a default thread pool from CamelContext#getErrorHandlerExecutorService
            executorService = null;
        }
    }
    return executorService;
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // create and register thread pool profile
            ThreadPoolProfile profile = new ThreadPoolProfile("myProfile");
            profile.setPoolSize(2);
            profile.setMaxPoolSize(8);
            profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);
            context.getExecutorServiceManager().registerThreadPoolProfile(profile);

            from("direct:start")
                .aggregate(header("id"), new BodyInAggregatingStrategy())
                    // use our custom thread pool profile
                    .completionSize(3).executorServiceRef("myProfile")
                    .to("log:foo")
                    .to("mock:aggregated");
        }
    };
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            ThreadPoolProfile profile = new ThreadPoolProfile("custom");
            profile.setPoolSize(5);
            profile.setMaxPoolSize(15);
            profile.setKeepAliveTime(25L);
            profile.setMaxQueueSize(250);
            profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);

            context.getExecutorServiceManager().registerThreadPoolProfile(profile);

            from("direct:start").threads().executorServiceRef("custom").to("mock:result");

            from("direct:foo").threads().executorServiceRef("custom").to("mock:foo");
        }
    };
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            ThreadPoolProfile profile = new ThreadPoolProfile("custom");
            profile.setPoolSize(5);
            profile.setMaxPoolSize(15);
            profile.setKeepAliveTime(25L);
            profile.setMaxQueueSize(250);
            profile.setAllowCoreThreadTimeOut(true);
            profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);

            context.getExecutorServiceManager().registerThreadPoolProfile(profile);

            from("direct:start").threads().executorServiceRef("custom").to("mock:result");
        }
    };
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testDefaultUnboundedQueueThreadPool() throws Exception {
    ThreadPoolProfile custom = new ThreadPoolProfile("custom");
    custom.setPoolSize(10);
    custom.setMaxPoolSize(30);
    custom.setKeepAliveTime(50L);
    custom.setMaxQueueSize(Integer.MAX_VALUE);

    context.getExecutorServiceManager().setDefaultThreadPoolProfile(custom);
    assertEquals(true, custom.isDefaultProfile().booleanValue());

    ExecutorService myPool = context.getExecutorServiceManager().newDefaultThreadPool(this, "myPool");
    assertEquals(false, myPool.isShutdown());

    // should use default settings
    ThreadPoolExecutor executor = (ThreadPoolExecutor) myPool;
    assertEquals(10, executor.getCorePoolSize());
    assertEquals(30, executor.getMaximumPoolSize());
    assertEquals(50, executor.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals(Integer.MAX_VALUE, executor.getQueue().remainingCapacity());

    context.stop();
    assertEquals(true, myPool.isShutdown());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testDefaultNoMaxQueueThreadPool() throws Exception {
    ThreadPoolProfile custom = new ThreadPoolProfile("custom");
    custom.setPoolSize(10);
    custom.setMaxPoolSize(30);
    custom.setKeepAliveTime(50L);
    custom.setMaxQueueSize(0);

    context.getExecutorServiceManager().setDefaultThreadPoolProfile(custom);
    assertEquals(true, custom.isDefaultProfile().booleanValue());

    ExecutorService myPool = context.getExecutorServiceManager().newDefaultThreadPool(this, "myPool");
    assertEquals(false, myPool.isShutdown());

    // should use default settings
    ThreadPoolExecutor executor = (ThreadPoolExecutor) myPool;
    assertEquals(10, executor.getCorePoolSize());
    assertEquals(30, executor.getMaximumPoolSize());
    assertEquals(50, executor.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals(0, executor.getQueue().remainingCapacity());

    context.stop();
    assertEquals(true, myPool.isShutdown());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testCustomDefaultThreadPool() throws Exception {
    ThreadPoolProfile custom = new ThreadPoolProfile("custom");
    custom.setKeepAliveTime(20L);
    custom.setMaxPoolSize(40);
    custom.setPoolSize(5);
    custom.setMaxQueueSize(2000);

    context.getExecutorServiceManager().setDefaultThreadPoolProfile(custom);
    assertEquals(true, custom.isDefaultProfile().booleanValue());

    ExecutorService myPool = context.getExecutorServiceManager().newDefaultThreadPool(this, "myPool");
    assertEquals(false, myPool.isShutdown());

    // should use default settings
    ThreadPoolExecutor executor = (ThreadPoolExecutor) myPool;
    assertEquals(5, executor.getCorePoolSize());
    assertEquals(40, executor.getMaximumPoolSize());
    assertEquals(20, executor.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals(2000, executor.getQueue().remainingCapacity());

    context.stop();
    assertEquals(true, myPool.isShutdown());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testTwoGetThreadPoolProfile() throws Exception {
    assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));

    ThreadPoolProfile foo = new ThreadPoolProfile("foo");
    foo.setKeepAliveTime(20L);
    foo.setMaxPoolSize(40);
    foo.setPoolSize(5);
    foo.setMaxQueueSize(2000);

    context.getExecutorServiceManager().registerThreadPoolProfile(foo);

    ThreadPoolProfile bar = new ThreadPoolProfile("bar");
    bar.setKeepAliveTime(40L);
    bar.setMaxPoolSize(5);
    bar.setPoolSize(1);
    bar.setMaxQueueSize(100);

    context.getExecutorServiceManager().registerThreadPoolProfile(bar);

    assertSame(foo, context.getExecutorServiceManager().getThreadPoolProfile("foo"));
    assertSame(bar, context.getExecutorServiceManager().getThreadPoolProfile("bar"));
    assertNotSame(foo, bar);

    assertFalse(context.getExecutorServiceManager().getThreadPoolProfile("foo").isDefaultProfile());
    assertFalse(context.getExecutorServiceManager().getThreadPoolProfile("bar").isDefaultProfile());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testGetThreadPoolProfileInheritCustomDefaultValues() throws Exception {
    ThreadPoolProfile newDefault = new ThreadPoolProfile("newDefault");
    newDefault.setKeepAliveTime(30L);
    newDefault.setMaxPoolSize(50);
    newDefault.setPoolSize(5);
    newDefault.setMaxQueueSize(2000);
    newDefault.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);
    context.getExecutorServiceManager().setDefaultThreadPoolProfile(newDefault);

    assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));
    ThreadPoolProfile foo = new ThreadPoolProfile("foo");
    foo.setMaxPoolSize(25);
    foo.setPoolSize(1);
    context.getExecutorServiceManager().registerThreadPoolProfile(foo);
    assertSame(foo, context.getExecutorServiceManager().getThreadPoolProfile("foo"));

    ExecutorService executor = context.getExecutorServiceManager().newThreadPool(this, "MyPool", "foo");

    ThreadPoolExecutor tp = assertIsInstanceOf(ThreadPoolExecutor.class, executor);
    assertEquals(25, tp.getMaximumPoolSize());
    // should inherit the default values
    assertEquals(1, tp.getCorePoolSize());
    assertEquals(30, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals("Abort", tp.getRejectedExecutionHandler().toString());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testGetThreadPoolProfileInheritCustomDefaultValues2() throws Exception {
    ThreadPoolProfile newDefault = new ThreadPoolProfile("newDefault");
    // just change the max pool as the default profile should then inherit the old default profile
    newDefault.setMaxPoolSize(50);
    context.getExecutorServiceManager().setDefaultThreadPoolProfile(newDefault);

    assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));
    ThreadPoolProfile foo = new ThreadPoolProfile("foo");
    foo.setPoolSize(1);
    context.getExecutorServiceManager().registerThreadPoolProfile(foo);
    assertSame(foo, context.getExecutorServiceManager().getThreadPoolProfile("foo"));

    ExecutorService executor = context.getExecutorServiceManager().newThreadPool(this, "MyPool", "foo");

    ThreadPoolExecutor tp = assertIsInstanceOf(ThreadPoolExecutor.class, executor);
    assertEquals(1, tp.getCorePoolSize());
    // should inherit the default values
    assertEquals(50, tp.getMaximumPoolSize());
    assertEquals(60, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals("CallerRuns", tp.getRejectedExecutionHandler().toString());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testNewThreadPoolProfile() throws Exception {
    assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));

    ThreadPoolProfile foo = new ThreadPoolProfile("foo");
    foo.setKeepAliveTime(20L);
    foo.setMaxPoolSize(40);
    foo.setPoolSize(5);
    foo.setMaxQueueSize(2000);

    ExecutorService pool = context.getExecutorServiceManager().newThreadPool(this, "Cool", foo);
    assertNotNull(pool);

    ThreadPoolExecutor tp = assertIsInstanceOf(ThreadPoolExecutor.class, pool);
    assertEquals(20, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals(40, tp.getMaximumPoolSize());
    assertEquals(5, tp.getCorePoolSize());
    assertFalse(tp.isShutdown());

    context.stop();

    assertTrue(tp.isShutdown());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testNewThreadPoolProfileById() throws Exception {
    assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));

    ThreadPoolProfile foo = new ThreadPoolProfile("foo");
    foo.setKeepAliveTime(20L);
    foo.setMaxPoolSize(40);
    foo.setPoolSize(5);
    foo.setMaxQueueSize(2000);

    context.getExecutorServiceManager().registerThreadPoolProfile(foo);

    ExecutorService pool = context.getExecutorServiceManager().newThreadPool(this, "Cool", "foo");
    assertNotNull(pool);

    ThreadPoolExecutor tp = assertIsInstanceOf(ThreadPoolExecutor.class, pool);
    assertEquals(20, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals(40, tp.getMaximumPoolSize());
    assertEquals(5, tp.getCorePoolSize());
    assertFalse(tp.isShutdown());

    context.stop();

    assertTrue(tp.isShutdown());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testNewScheduledThreadPoolProfileById() throws Exception {
    assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));

    ThreadPoolProfile foo = new ThreadPoolProfile("foo");
    foo.setKeepAliveTime(20L);
    foo.setMaxPoolSize(40);
    foo.setPoolSize(5);
    foo.setMaxQueueSize(2000);

    context.getExecutorServiceManager().registerThreadPoolProfile(foo);

    ExecutorService pool = context.getExecutorServiceManager().newScheduledThreadPool(this, "Cool", "foo");
    assertNotNull(pool);

    SizedScheduledExecutorService tp = assertIsInstanceOf(SizedScheduledExecutorService.class, pool);
    // a scheduled dont use keep alive
    assertEquals(0, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals(Integer.MAX_VALUE, tp.getMaximumPoolSize());
    assertEquals(5, tp.getCorePoolSize());
    assertFalse(tp.isShutdown());

    context.stop();

    assertTrue(tp.isShutdown());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testLowProfile() throws Exception {
    CamelContext context = getMandatoryBean(CamelContext.class, "camel-C");

    ThreadPoolProfile profile = context.getExecutorServiceManager().getThreadPoolProfile("low");
    assertEquals(1, profile.getPoolSize().intValue());
    assertEquals(5, profile.getMaxPoolSize().intValue());
    assertEquals(null, profile.getKeepAliveTime());
    assertEquals(null, profile.getMaxQueueSize());
    assertEquals(null, profile.getRejectedPolicy());

    // create a thread pool from low
    ExecutorService executor = context.getExecutorServiceManager().newThreadPool(this, "MyLow", "low");
    ThreadPoolExecutor tp = assertIsInstanceOf(ThreadPoolExecutor.class, executor);
    assertEquals(1, tp.getCorePoolSize());
    assertEquals(5, tp.getMaximumPoolSize());
    // should inherit default options
    assertEquals(60, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals("CallerRuns", tp.getRejectedExecutionHandler().toString());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
public void testBigProfile() throws Exception {
    CamelContext context = getMandatoryBean(CamelContext.class, "camel-C");

    ThreadPoolProfile profile = context.getExecutorServiceManager().getThreadPoolProfile("big");
    assertEquals(50, profile.getPoolSize().intValue());
    assertEquals(100, profile.getMaxPoolSize().intValue());
    assertEquals(ThreadPoolRejectedPolicy.DiscardOldest, profile.getRejectedPolicy());
    assertEquals(null, profile.getKeepAliveTime());
    assertEquals(null, profile.getMaxQueueSize());

    // create a thread pool from big
    ExecutorService executor = context.getExecutorServiceManager().newThreadPool(this, "MyBig", "big");
    ThreadPoolExecutor tp = assertIsInstanceOf(ThreadPoolExecutor.class, executor);
    assertEquals(50, tp.getCorePoolSize());
    assertEquals(100, tp.getMaximumPoolSize());
    // should inherit default options
    assertEquals(60, tp.getKeepAliveTime(TimeUnit.SECONDS));
    assertEquals("DiscardOldest", tp.getRejectedExecutionHandler().toString());
}
 

import org.apache.camel.spi.ThreadPoolProfile; //导入依赖的package包/类
protected static synchronized ExecutorService getExecutorService(CamelContext context) {
    // CamelContext will shutdown thread pool when it shutdown so we can
    // lazy create it on demand
    // but in case of hot-deploy or the likes we need to be able to
    // re-create it (its a shared static instance)
    if (executorService == null || executorService.isTerminated() || executorService.isShutdown()) {
        final ExecutorServiceManager manager = context.getExecutorServiceManager();

        // try to lookup a pool first based on profile
        ThreadPoolProfile poolProfile = manager.getThreadPoolProfile(
            FacebookConstants.FACEBOOK_THREAD_PROFILE_NAME);
        if (poolProfile == null) {
            poolProfile = manager.getDefaultThreadPoolProfile();
        }

        // create a new pool using the custom or default profile
        executorService = manager.newScheduledThreadPool(FacebookProducer.class,
            FacebookConstants.FACEBOOK_THREAD_PROFILE_NAME, poolProfile);
    }

    return executorService;
}