spring-integration - spring 集成任务执行器队列充满了更多记录

Question

我开始构建一个 Spring Integration 应用程序，其中输入网关生成固定数量 (50) 的记录，然后停止生成新记录。中间有基本的filters/routers/transformer，结束的service activator和task executor config如下:

<int:service-activator input-channel="inChannel" output-channel="outChannel" ref="svcProcessor">
        <int:poller fixed-rate="100" task-executor="myTaskExecutor"/>
</int:service-activator>

<task:executor id = "myTaskExecutor" pool-size="5" queue-capacity="100"/>

我试图在 svcProcessor 方法的开头放置一些调试信息:

@Qualifier(value="myTaskExecutor")
@Autowired 
ThreadPoolTaskExecutor executor;

@ServiceActivator
public Order processOrder(Order order) {
  log.debug("---- " + "executor size: " + executor.getActiveCount() + 
        " q: " + executor.getThreadPoolExecutor().getQueue().size() + 
        " r: " + executor.getThreadPoolExecutor().getQueue().remainingCapacity()+
        " done: " + executor.getThreadPoolExecutor().getCompletedTaskCount() +
        " task: " + executor.getThreadPoolExecutor().getTaskCount()
    );
   //
   //process order takes up to 5 seconds.
   //
   return order;
}

有时程序运行后，日志显示队列已超过50，然后最终得到拒绝异常:

23:38:31.096 DEBUG [myTaskExecutor-2] ---- executor size: 5 q: 44 r: 56 done: 11 task: 60
23:38:31.870 DEBUG [myTaskExecutor-5] ---- executor size: 5 q: 51 r: 49 done: 11 task: 67
23:38:33.600 DEBUG [myTaskExecutor-4] ---- executor size: 5 q: 69 r: 31 done: 11 task: 85
23:32:46.792 DEBUG [myTaskExecutor-1] ---- executor size: 5 q: 72 r: 28 done: 11 task: 88

看起来active count和queue size/remaining的总和在5和100的配置下看起来是对的，但我不明白为什么队列中有超过50条记录，而且taskCount也比限制 50 个。

我是否查看了来自执行程序和队列的错误信息？

谢谢

更新:(不确定我是否应该打开另一个问题)

我尝试了 spring-integration(分支 SI3.0.x)的 cafeDemo 的 xml 版本，并使用了文档中提供的池，但使用了 100 毫秒的速率并增加了容量:

<int:service-activator input-channel="hotDrinks" ref="barista" method="prepareHotDrink" output-channel="preparedDrinks">
    <int:poller task-executor="pool" fixed-rate="100"/>          
</int:service-activator>
<task:executor id="pool" pool-size="5" queue-capacity="200"/>

我运行之后，也是在20次左右投递后出现rejection异常:

 org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@6c31732b[Running, pool size = 5, active threads = 5, queued tasks = 200, completed tasks = 0]]

在出现异常之前只有大约 32 个订单，所以我不确定为什么排队任务 = 200 而完成任务 = 0？

谢谢

Answer 1

getTaskCount() 此方法给出自启动以来分配给执行程序的任务总数。因此，它会随着时间的推移而增加。

其他变量是近似值，并不完全符合 java 文档。

1. getCompletedTaskCount()返回已完成执行的任务的大概总数。
2. public int getActiveCount()返回正在执行任务的线程的大概数量。

理想情况下，getTaskCount() 和 getCompletedTaskCount() 将随时间线性增加，因为它包括自代码开始执行以来分配的所有先前任务。然而，activeCount 应该小于 50，但作为近似数，有时会超过 50，余量很小。

引用:- https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html