Java 类org.springframework.core.task.SimpleAsyncTaskExecutor 实例源码

/**
 * 根据文件访问URL列表, 将文件从云存储或应用系统Context路径下的文件删除
 * <p>
 * 调用带有返回值的多线程（实现callable接口），也是同步的。参考：http://blueram.iteye.com/blog/1583117
 *
 * @param fileUrls
 * @return 返回存储路径
 */
public Integer deleteFiles(Collection<String> fileUrls) {
    int count = 0;
    AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
    for (String url : fileUrls) {
        final String fileUrl = StringUtils.substringAfterLast(url, coreConfig.getValue("bae.bcs.bucket") + "/");
        try {
            Future<Integer> future = executor.submit(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    deleteFile(fileUrl);
                    return 1;
                }
            });
            count += future.get();
        } catch (InterruptedException | ExecutionException e) {
            Exceptions.printException(e);
        }
    }
    return count;
}

@Bean
    public TaskExecutor taskExecutor() {
        //https://jira.spring.io/browse/BATCH-2269
        final SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor() {
            @Override
            protected void doExecute(Runnable task) {
                //gets the jobExecution of the configuration thread
                final JobExecution jobExecution = JobSynchronizationManager.getContext().getJobExecution();
                super.doExecute(() -> {
                    JobSynchronizationManager.register(jobExecution);
                    try {
                        task.run();
                    } finally {
//                        JobSynchronizationManager.release();
                        JobSynchronizationManager.close();
                    }
                });
            }
        };
        simpleAsyncTaskExecutor.setConcurrencyLimit(20);
        return simpleAsyncTaskExecutor;
    }

@Override
public Executor getAsyncExecutor() {
    if (properties.isEnabled()) {
        ThreadPoolTaskExecutor executor = null;
        try {
            executor = beanFactory.getBean(ThreadPoolTaskExecutor.class);
        } catch (NoUniqueBeanDefinitionException e) {
            executor = beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, ThreadPoolTaskExecutor.class);
        } catch (NoSuchBeanDefinitionException ex) {
        }
        if (executor != null) {
            log.info("use default TaskExecutor ...");
            return executor;
        } else {
            throw new BeanCreationException("Expecting a 'ThreadPoolTaskExecutor' exists, but was 'null'");
        }
    } else {
        log.info(
                "'AsyncExecutorConfiguration' is disabled, so create 'SimpleAsyncTaskExecutor' with 'threadNamePrefix' - '{}'",
                properties.getThreadNamePrefix());
        return new SimpleAsyncTaskExecutor(properties.getThreadNamePrefix());
    }
}

@Before
public void setUp() {
    ReflectionTestUtils.setField(
            instance2Test, 
            "hwAdapter_Rest", 
            mk_hwAdapter); 

    //~ INIT ONCE ~~~~~~~~~~~~~~
    if(initialized)return ;
    initialized = true;
    ReflectionTestUtils.setField(
            instance2Test, 
            "asyncExecutor", 
            new SimpleAsyncTaskExecutor("testBulbsHwService_")); 

    ReflectionTestUtils.setField(new BulbsCoreEventProcessor(), "eventStore", 
            mk_eventStore);
    serviceLocator = new DomainServiceLocator();
    ReflectionTestUtils.setField(serviceLocator, "instance", mk_domainServiceLocator);
}

/**
 * Create the task executor which will be used for multi-threading
 *
 * @return TaskExecutor
 */
@Bean
public TaskExecutor taskExecutor() {
    SimpleAsyncTaskExecutor asyncTaskExecutor = new SimpleAsyncTaskExecutor("spring_batch");
    asyncTaskExecutor.setConcurrencyLimit(SimpleAsyncTaskExecutor.NO_CONCURRENCY);
    return asyncTaskExecutor;
}

@Test
public void taskExecutor() throws Exception {

    URI uri = new URI("ws://localhost/abc");
    this.wsClient.setTaskExecutor(new SimpleAsyncTaskExecutor());
    WebSocketSession session = this.wsClient.doHandshake(this.wsHandler, this.headers, uri).get();

    assertNotNull(session);
}

@Test
public void doHandshakeWithTaskExecutor() throws Exception {

    WebSocketHttpHeaders headers = new WebSocketHttpHeaders();
    headers.setSecWebSocketProtocol(Arrays.asList("echo"));

    this.client.setTaskExecutor(new SimpleAsyncTaskExecutor());
    this.wsSession = this.client.doHandshake(new TextWebSocketHandler(), headers, new URI(this.wsUrl)).get();

    assertEquals(this.wsUrl, this.wsSession.getUri().toString());
    assertEquals("echo", this.wsSession.getAcceptedProtocol());
}

@Override
protected AsyncClientHttpRequestFactory createRequestFactory() {
    SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
    AsyncListenableTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    requestFactory.setTaskExecutor(taskExecutor);
    return requestFactory;
}

@Bean
public Step step1() {
    return stepBuilderFactory.get("step1")
            .<Bracket, Bracket>chunk(10000)
            .reader(itemReader())
            .processor(itemProcessor())
            .writer(itemWriter(null))
            .taskExecutor(new SimpleAsyncTaskExecutor())
            .build();
}

@Before
public void setup() {
    executor = new SimpleAsyncTaskExecutor();
    recorder = LogbackRecorder.forClass(MockEnvironment.class).reset().capture("ALL");
    environment = new MockEnvironment();
    recorder.release();
    config = spy(new TestAsyncSpringLiquibase(executor, environment));
    recorder = LogbackRecorder.forClass(AsyncSpringLiquibase.class).reset().capture("ALL");
}

/**
 * @return task executor
 *
 * @see https://jira.spring.io/browse/BATCH-2269
 */
@Bean
public TaskExecutor taskExecutor() {
    return new SimpleAsyncTaskExecutor() {
        private static final long serialVersionUID = 1L;

        @Override
        protected void doExecute(Runnable task) {

            /* Gets the jobExecution of the configuration thread */
            JobExecution jobExecution = JobSynchronizationManager
                    .getContext()
                    .getJobExecution();

            super.doExecute(new Runnable() {
                @Override
                public void run() {
                    JobSynchronizationManager.register(jobExecution);

                    try {
                        task.run();
                    } finally {
                        JobSynchronizationManager.release();
                    }
                }
            });
        }
    };
}

/**
 * Initializes the task executor adapter. Expects the delegate to be set. If no
 * {@link #setTaskExecutor(org.springframework.core.task.TaskExecutor) taskExecutor} is
 * provided will create a default one using {@link org.springframework.core.task.SimpleAsyncTaskExecutor}. 
 */
public void afterPropertiesSet() throws Exception {
    Assert.notNull(delegate, "delegate SpaceDataEventListener must not be null");
    if (taskExecutor == null) {
        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
        simpleAsyncTaskExecutor.setDaemon(true);
        taskExecutor = simpleAsyncTaskExecutor;
    }
}

@Bean(name = "networkPool")
public SimpleAsyncTaskExecutor networkPool()
{
  SimpleAsyncTaskExecutor pool = new SimpleAsyncTaskExecutor();
  pool.setThreadPriority(Thread.NORM_PRIORITY + 1);
  return pool;
}

private AsyncRestTemplate getAsyncRestTemplate() {
    SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
    factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
    factory.setConnectTimeout(gorouterProperties.getConnectTimeout());
    factory.setReadTimeout(gorouterProperties.getReadTimeout());

    return new AsyncRestTemplate(factory);
}

public CSVSerializerTest() {
    this.serializer = new CSVSerializer();
    TaskExecutor executor = new SimpleAsyncTaskExecutor();
    ReflectionTestUtils.setField(serializer, "executor", executor);
    ReflectionTestUtils.setField(serializer, "defaultTextEnclosure", "\"");
    ReflectionTestUtils.setField(serializer, "defaultEscapeChar", "\u0000");

}

@Test
public void testCustomTaskExecutor() throws Exception {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    container.setTaskExecutor(taskExecutor);

    container.setAmazonSqs(mock(AmazonSQSAsync.class, withSettings().stubOnly()));
    container.setMessageHandler(mock(QueueMessageHandler.class));
    container.setBeanName("testContainerName");
    container.afterPropertiesSet();

    assertEquals(taskExecutor, container.getTaskExecutor());
}

@Test
public void testContainerDoesNotProcessMessageAfterBeingStopped() throws Exception {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    container.setTaskExecutor(taskExecutor);

    AmazonSQSAsync sqs = mock(AmazonSQSAsync.class, withSettings().stubOnly());
    container.setAmazonSqs(sqs);

    QueueMessageHandler messageHandler = new QueueMessageHandler() {

        @Override
        public void handleMessage(org.springframework.messaging.Message<?> message) throws MessagingException {
            fail("Should not have been called");
        }
    };
    container.setMessageHandler(messageHandler);
    container.setBeanName("testContainerName");

    mockGetQueueUrl(sqs, "testQueue", "http://testContainerDoesNotProcessMessageAfterBeingStopped.amazonaws.com");

    container.afterPropertiesSet();

    when(sqs.receiveMessage(new ReceiveMessageRequest("http://testContainerDoesNotProcessMessageAfterBeingStopped.amazonaws.com"))).
            thenAnswer((Answer<ReceiveMessageResult>) invocation -> {
                container.stop();
                return new ReceiveMessageResult().withMessages(new Message().withBody("test"));
            });

    container.start();
}

/** job launcher used for executing jobs on http requests, here: for batch admin */
@Bean 
public JobLauncher asyncJobLauncher() throws Exception {
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    jobLauncher.setJobRepository(jobRepository());
    jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());
    jobLauncher.afterPropertiesSet();
    return jobLauncher;
}

/**
 * {@inheritDoc}
 * 
 * Cleanup the configuration items.
 */
public void destroy() {

    synchronized (lock) {
        if (isMulticasterManagedInternally) {
            eventMulticaster.removeAllListeners();
            eventMulticaster = null;
        }

        if (extenderConfiguration != null) {
            extenderConfiguration.close();
            extenderConfiguration = null;
        }

        // postpone the task executor shutdown
        if (forceThreadShutdown) {

            if (isTaskExecutorManagedInternally) {
                log.warn("Forcing the (internally created) taskExecutor to stop...");
                ThreadGroup th = ((SimpleAsyncTaskExecutor) taskExecutor).getThreadGroup();
                if (!th.isDestroyed()) {
                    // ask the threads nicely to stop
                    th.interrupt();
                }
            }
            taskExecutor = null;
        }

        if (isShutdownTaskExecutorManagedInternally) {
            try {
                ((DisposableBean) shutdownTaskExecutor).destroy();
            }
            catch (Exception ex) {
                log.debug("Received exception while shutting down shutdown task executor", ex);
            }
            shutdownTaskExecutor = null;
        }
    }
}

private TaskExecutor createDefaultTaskExecutor() {
    // create thread-pool for starting contexts
    ThreadGroup threadGroup = new ThreadGroup("spring-osgi-extender[" + ObjectUtils.getIdentityHexString(this)
            + "]-threads");
    threadGroup.setDaemon(false);

    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    taskExecutor.setThreadGroup(threadGroup);
    taskExecutor.setThreadNamePrefix("SpringOsgiExtenderThread-");

    isTaskExecutorManagedInternally = true;

    return taskExecutor;
}

private JobLauncher createJobLauncher() throws Exception {
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    jobLauncher.setJobRepository(jobRepository);
    jobLauncher.setTaskExecutor( new SimpleAsyncTaskExecutor());
    jobLauncher.afterPropertiesSet();
    return jobLauncher;
}

/**
 * Create a new {@code AsyncAnnotationAdvisor} for bean-style configuration.
 */
public AsyncAnnotationAdvisor() {
    this(new SimpleAsyncTaskExecutor());
}

@Override
public Executor getAsyncExecutor() {
  return new TracedExecutor(new SimpleAsyncTaskExecutor(), tracer);
}

@Bean
public Executor simpleAsyncTaskExecutor() {
  return new SimpleAsyncTaskExecutor();
}

@Bean(name="taskScheduler")
public Executor taskScheduler() {
    return new SimpleAsyncTaskExecutor();
}

@Bean
public TaskExecutor consumerTaskExecutor() {
    return new SimpleAsyncTaskExecutor();
}

@Bean
public TaskExecutor consumerTaskExecutor() {
    return new SimpleAsyncTaskExecutor();
}

@Bean
public SimpleApplicationEventMulticaster applicationEventMulticaster() {
    final SimpleApplicationEventMulticaster multicaster = new SimpleApplicationEventMulticaster();
    multicaster.setTaskExecutor(new SimpleAsyncTaskExecutor());
    return multicaster;
}

@Bean
public ExecutorChannel outboundRequests() {
    return MessageChannels.executor("outboundRequests", new SimpleAsyncTaskExecutor()).get();
}

/**
 * Create a new {@code AsyncAnnotationAdvisor} for bean-style configuration.
 */
public AsyncAnnotationAdvisor() {
    this(new SimpleAsyncTaskExecutor());
}

@Conditional(RestifyAsyncConfiguration.RestifyAsyncTaskExecutorCondition.class)
@Bean
public AsyncListenableTaskExecutor restifyAsyncTaskExecutor() {
    return new SimpleAsyncTaskExecutor("RestifyAsyncTaskExecutor");
}

public WebAsyncTaskEndpointCallExecutableFactory() {
    this(new SimpleAsyncTaskExecutor());
}

public WebAsyncTaskEndpointCallExecutableFactory(Long timeout) {
    this(timeout, new SimpleAsyncTaskExecutor());
}

public ListenableFutureTaskEndpointCallExecutableFactory() {
    this(new SimpleAsyncTaskExecutor());
}

public ListenableFutureEndpointCallExecutableFactory() {
    this(new SimpleAsyncTaskExecutor("ListenableFutureEndpointCallExecutable"));
}

@Bean
public TaskExecutor taskExecutor() {
    return new SimpleAsyncTaskExecutor();
}

@Bean(name = "applicationEventMulticaster")
public ApplicationEventMulticaster simpleApplicationEventMulticaster() {
  SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();
  eventMulticaster.setTaskExecutor(new SimpleAsyncTaskExecutor());
  return eventMulticaster;
}

@Bean(name = "networkTaskExecutor")
public SimpleAsyncTaskExecutor roundPool()
{
  SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
  return taskExecutor;
}

@Override
protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group,
        ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
    Assert.state(!HeaderMode.embeddedHeaders.equals(properties.getHeaderMode()),
            "the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively");
    String destination = consumerDestination.getName();
    SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
            this.connectionFactory);
    listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
    listenerContainer.setChannelTransacted(properties.getExtension().isTransacted());
    listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
    int concurrency = properties.getConcurrency();
    concurrency = concurrency > 0 ? concurrency : 1;
    listenerContainer.setConcurrentConsumers(concurrency);
    int maxConcurrency = properties.getExtension().getMaxConcurrency();
    if (maxConcurrency > concurrency) {
        listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
    }
    listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
    listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
    listenerContainer.setTxSize(properties.getExtension().getTxSize());
    listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
    listenerContainer.setQueueNames(consumerDestination.getName());
    listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
    listenerContainer.setMessagePropertiesConverter(
            RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
    listenerContainer.setExclusive(properties.getExtension().isExclusive());
    listenerContainer.setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal());
    if (properties.getExtension().getQueueDeclarationRetries() != null) {
        listenerContainer.setDeclarationRetries(properties.getExtension().getQueueDeclarationRetries());
    }
    if (properties.getExtension().getFailedDeclarationRetryInterval() != null) {
        listenerContainer.setFailedDeclarationRetryInterval(
                properties.getExtension().getFailedDeclarationRetryInterval());
    }
    listenerContainer.afterPropertiesSet();

    AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
    adapter.setBeanFactory(this.getBeanFactory());
    adapter.setBeanName("inbound." + destination);
    DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
    mapper.setRequestHeaderNames(properties.getExtension().getHeaderPatterns());
    adapter.setHeaderMapper(mapper);
    ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(consumerDestination, group, properties);
    if (properties.getMaxAttempts() > 1) {
        adapter.setRetryTemplate(buildRetryTemplate(properties));
        if (properties.getExtension().isRepublishToDlq()) {
            adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
        }
    }
    else {
        adapter.setErrorMessageStrategy(errorMessageStrategy);
        adapter.setErrorChannel(errorInfrastructure.getErrorChannel());
    }
    return adapter;
}

@Bean
public AsyncTaskExecutor asyncTaskExecutor() {
    return new SimpleAsyncTaskExecutor("async");
}