在使用任务执行器时，在LinkedBlockingQueue和SynchronousQueue之间进行选择。

huangapple
未分类评论45阅读模式
英文:

Selecting between task executor with LinkedBlockingQueue and SynchronousQueue

问题

我目前有一个处理器实现，它使用LinkedBlockingQueue创建一个ThreadPoolExecutor，以提供对传入作业/任务的异步提交。这样做的目的是为客户提供即时响应并返回一些初始响应。

下面的TradeProcessor位于一个公共项目中（假设为项目A），并被其他项目（假设项目B和C）使用：

  1. public class TradeProcessor implements Processor {
  3.     private final TestRepository repo;
  4.     private final JobRunner runner;
  6.     private int corePoolSize = 1; //始终为1
  7.     private int maxPoolSize = 1; //始终为1
  8.     private int keepAliveTime = 60;
  9.     private int queueSize = 10;
  11.     private ExecutorService executorService;
  13.     public AsyncProcessor(final TestRepository repo, final JobRunner runner) {
  14.         this.repo = repo;
  15.         this.runner = runner;
  16.     }
  18.     @PostConstruct
  19.     public void init() {       
  20.         executorService = createNewThreadPoolExecutor();
  21.     }
  23.     @Override
  24.     public MyResponse process(final TestRequest request, final String id) {
  25.       
  26.         executorService.execute(new JobRunTask(runner, repo, Pair.of(id, request)));
  28.         return new MyResponse(request.getName(), id, SUBMITTED, initialJobStatus());
  29.     }
  31.     private ThreadPoolExecutor createNewThreadPoolExecutor() {
  32.         return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, SECONDS,
  33.                                       new LinkedBlockingQueue<> (queueSize),
  34.                                       new ThreadFactoryBuilder().setNameFormat("abc-thread-%d").build(),
  35.                                       createRejectedExecutionHandler()) {
  36.             @Override
  37.             protected void afterExecute(Runnable r, Throwable t) {
  38.                 if (null != t) {
  39.                    //bla bla
  40.                 }
  41.             }
  42.         };
  43.     }
  45.     private RejectedExecutionHandler createRejectedExecutionHandler() {
  46.         return (runnable, executor) -> {
  47. 			//bla bla
  48.         };
  49.     }
  50. }

我现在希望在上述处理器中添加一些同步功能。例如，某个项目可以选择“等待”响应，并获得实际的完整响应，而不是初始响应。当任务执行/提交并获得结果后，它会阻塞并返回响应。
我理解如果我使用SynchronousQueue创建一个ThreadPoolExecutor，可以满足这个需求。如何实现这一点，并根据各个项目提供的一些标志在同步和异步之间进行切换呢？

英文:

I currently have a processor implementation that is creating a ThreadPoolExecutor using a LinkedBlockingQueue as to offer async submissions of incoming jobs/tasks. The intention for this is to offer client an immediate response with some initial response.

The below TradeProcessor sits in a common project (say Project A) and is used by other projects (say project B and C):

  1. public class TradeProcessor implements Processor {
  3.     private final TestRepository repo;
  4.     private final JobRunner runner;
  6.     private int corePoolSize = 1 //always 1
  7.     private int maxPoolSize = 1; //always 1
  8.     private int keepAliveTime = 60
  9.     private int queueSize = 10;
  11.     private ExecutorService executorService;
  13.     public AsyncProcessor(final TestRepository repo, final JobRunner runner) {
  14.         this.repo = repo;
  15.         this.runner = runner;
  16.     }
  18.     @PostConstruct
  19.     public void init() {       
  20.         executorService = createNewThreadPoolExecutor();
  21.     }
  23.     @Override
  24.     public MyResponse process(final TestRequest request, final String id) {
  25.       
  26.         executorService.execute(new JobRunTask(runner, repo, Pair.of(id, request)));
  28.         return new MyResponse(request.getName(), id, SUBMITTED, initialJobStatus());
  29.     }
  31.     private ThreadPoolExecutor createNewThreadPoolExecutor() {
  32.         return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, SECONDS,
  33.                                       new LinkedBlockingQueue&lt;&gt;(queueSize),
  34.                                       new ThreadFactoryBuilder().setNameFormat(&quot;abc-thread-%d&quot;).build(),
  35.                                       createRejectedExecutionHandler()) {
  36.             @Override
  37.             protected void afterExecute(Runnable r, Throwable t) {
  38.                 if (null != t) {
  39.                    //bla bla
  40.                 }
  41.             }
  42.         };
  43.     }
  45.     private RejectedExecutionHandler createRejectedExecutionHandler() {
  46.         return (runnable, executor) -&gt; {
  47. 			//bla bla
  48.         };
  49.     }
  50. }

I am now looking to add some synchronous functionality to the above Processor. For example, A project may choose that it can wait for a response and have the actual full response back and not the initial one. It blocks when task executed/submitted and has the result back and sends response.
I understand I can satisfy this requirement if I create an ThreadPoolExecutor using SynchronousQueue? How can I implement this and switch between the two modes, sync and async, based on some flags that the individuals projects will provide?

答案1

得分: 0

你不需要使用SynchronousQueue，可以直接调用JobRunTask的方法，在你想要同步发生的时候获取响应。

Processor接口定义了三个方法：

  1. interface Processor {
  2.     // 这将将请求放入队列并返回默认值
  3.     MyResponse process(final TestRequest request, final String id);
  4.     // 这将同步处理请求
  5.     MyResponse processSync(final TestRequest request, final String id);
  6.     // 这将返回一个Future，用于检查执行是否完成，同时返回一个响应
  7.     Future<MyResponse> processAsync(final TestRequest request, final String id);
  8. }

你的JobRunTask类将定义两个公共方法，与只有一个方法相反，在其中一个方法中，它将通过执行业务逻辑返回响应，而在另一个方法中，它将作为可运行任务运行。

  1. class JobRunTask implements Runnable {
  2.     @Override
  3.     public void run() {
  4.         MyResponse response = execute();
  5.         // 对响应进行处理
  6.     }
  8.     public MyResponse execute() {
  9.         // 你的业务逻辑
  10.         return null;
  11.     }
  12. }

一旦你掌握了这两个接口，你可以实现你的TradeProcessor类如下：

  1. public class TradeProcessor implements Processor {
  2.     private ExecutorService executorService;
  3.     private final JobRunner runner;
  4.     private final TestRepository repo;
  6.     @Override
  7.     public MyResponse process(final TestRequest request, final String id) {
  8.         executorService.execute(new JobRunTask(runner, repo, Pair.of(id, request)));
  9.         return new MyResponse(request.getName(), id, SUBMITTED, initialJobStatus());
  10.     }
  12.     @Override
  13.     public MyResponse processSync(final TestRequest request, final String id) {
  14.         return new JobRunTask(runner, repo, Pair.of(id, request)).execute();
  15.     }
  17.     @Override
  18.     public Future<MyResponse> processAsync(final TestRequest request, final String id) {
  19.         return executorService.submit(() -> new JobRunTask(runner, repo, Pair.of(id, request)));
  20.     }
  21. }

其他初始化/对象创建等细节留给操作者处理。

使用这种方法，你不需要在不同的队列之间切换。它还允许调用者查看请求是否已完成，在调用者并行运行多个线程的情况下。

英文:

You do not have to use SynchrnousQueue, you can directly call the method of the JobRunTask to get the response when you want things to happen synchronously.

Processor interface defines three method

  1. interface Processor {
  2.     // This will enqueue the request and returns default value
  3.     MyResponse process(final TestRequest request, final String id);
  4.     // this will process request in sync
  5.     MyResponse processSync(final TestRequest request, final String id);
  6.     // this returns a future that can be used to check if the execution is completed or not and also returns a response.
  7.     Future&lt;MyResponse&gt; processAsync(final TestRequest request, final String id);
  8.   }

Your JobRunTask class would define two public methods as opposed to one method, in one of the method it will return the response by executing the business logic while in other it will run as a runnable task.

  1.   class JobRunTask implements Runnable {
  2.     @Override
  3.     public void run() {
  4.       MyResponse response = execute();
  5.       // do something with response
  6.     }
  8.     public MyResponse execute() {
  9.       // your business logic
  10.       return null;
  11.     }
  12.   }

Once you have these two interfaces handy you can implement your TradeProcessor class as.

  1.   public class TradeProcessor implements Processor {
  2.     private ExecutorService executorService;
  3.     private final JobRunner runner;
  4.     private final TestRepository repo;
  6.     @Override
  7.     public MyResponse process(final TestRequest request, final String id) {
  8.       executorService.execute(new JobRunTask(runner, repo, Pair.of(id, request)));
  9.       return new MyResponse(request.getName(), id, SUBMITTED, initialJobStatus());
  10.     }
  12.     @Override
  13.     public MyResponse processSync(final TestRequest request, final String id) {
  14.       return new JobRunTask(runner, repo, Pair.of(id, request)).execute();
  15.     }
  17.     @Override
  18.     public Future&lt;MyResponse&gt; processAsync(final TestRequest request, final String id) {
  19.       return executorService.submit(() -&gt; new JobRunTask(runner, repo, Pair.of(id, request)));
  20.     }
  21.   }

Other details like initialization/object creations etc have been left for the op.

Using this you would not have to switch from one queue to another. It will also allow the caller to see the request was completed or not, in case the caller runs multiple threads in parallel.

huangapple
  • 本文由 发表于 2020年7月23日 16:53:33
  • 转载请务必保留本文链接：https://java.coder-hub.com/63050476.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定