spring - blockingqueue 和 Spring - 如何在启动时启动线程池？

Question

我有一个使用阻塞队列来运行生产者-消费者设计的 Spring 应用程序。基本上，当用户使用 REST Controller 进行 API 调用时，它会向阻塞队列添加一个工作，并且后台线程将在队列到达后立即使用该队列。

我看到Spring建议使用它的TaskExecutor，所以我有以下类(class)。
线程配置.java

@Configuration
public class ThreadConfig {

  @Bean
  public TaskExecutor threadPoolTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4);
    executor.setMaxPoolSize(4);
    executor.setThreadNamePrefix("default_task_executor_thread");
    executor.initialize();
    return executor;
  }

}

我还有一个消费者组件，它监视队列并运行任务。
消息消费者.java

@Component
public class MessageConsumer implements Runnable{


  private final BlockingQueue<String> queue;

  MessageConsumer(BlockingQueue<String> queue){
    this.queue = queue;
  }

  public void run(){
    try {
      while (true) {
        String str = queue.take();
        // Do something
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

}

现在，我不确定如何在 spring 应用程序启动时启动线程池。

我是否只是在 Main 中添加代码？

任何帮助，将不胜感激。谢谢

Answer 1

我觉得你想多了。内部 ThreadPoolTaskExecutor也使用 BlockingQueue把工作交给 worker Threads来自线程池。由于您没有更改默认值queueCapacity它将是 LinkedBlockingQueue .

在您的场景中，这样做会更容易:

因为每个 HTTP 请求都由单独的 Thread 处理，生产者将是 Thread处理 HTTP 请求的

消费者将是 worker Thread来自 ThreadPoolTaskExecutor .

因此，要使生产者-消费者工作，只需使用 ThreadPoolTaskExecutor只需创建一个任务并将其提交到您的 ThreadPool，其中一名工作人员将使用它:

@RestController
public class MyController {

    private final TaskExecutor taskExecutor;

    @Autowired
    public MyController(final TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    @GetMapping("/test/{value}")
    public ResponseEntity<String> get(final @PathVariable("value") String value) {
        taskExecutor.execute(() -> {

            System.out.println(value);
            // do something with your String
            // this will be executed by some worker Thread
        });

        return ResponseEntity.ok(value);
    }
}

而且由于您使用的是 ThreadPoolTaskExecutor它实现了 DisposableBean接口(interface)(通过扩展 ExecutorConfigurationSupport ) - 你不必 shutdown池明确。当 Spring Context 将被销毁并且 destroy 时，Spring 将为您执行此操作该接口(interface)的方法将在池 Bean 上调用。