Spring 与多个执行器通道的集成不并行处理

Question

我有一个要求，我需要将消息传递到多个通道 asyc。为了使我的流程异步，我正在使用所有执行程序通道。但由于某种原因，流程仍然是顺序的。我可以看到我在任务执行器中配置的差异线程，但按顺序排列。 这是我正在使用的配置

<int:channel id="mainChannel">
        <int:interceptors>
            <int:wire-tap channel="channel1"/>
            <int:wire-tap channel="channel2"/>
            <int:wire-tap channel="channel3"/>
      
</int:interceptors>
    </int:channel> 
   <int:channel id="channel1">
        <int:dispatcher task-executor="exec1" />
   </int:channel>
  <int:channel id="channel2">
        <int:dispatcher task-executor="exec2" />
  </int:channel>
   <int:channel id="channel3">
        <int:dispatcher task-executor="exec3" />
  </int:channel>

据我了解，所有这些都是异步的（在我的情况下，3 个线程应该并行运行）

从日志中我可以看到所有顺序但具有不同的线程名称.. 我假设 preSend/Postsend 应该以随机顺序调用。

我是否缺少任何东西来并行创建多个执行器通道。

我将非常感谢任何帮助。

Answer 1

您可能需要调用异步实现 bean，如下所示：

<beans:bean id="asyncExecutor"
            class="org.springframework.core.task.SimpleAsyncTaskExecutor"/>

---

   <int:channel id="channel1">
        <int:dispatcher task-executor="asyncExecutor" />
   </int:channel>
  <int:channel id="channel2">
        <int:dispatcher task-executor="asyncExecutor" />
  </int:channel>
   <int:channel id="channel3">
        <int:dispatcher task-executor="asyncExecutor" />
  </int:channel>

SimpleAsyncTaskExecutor 说明：

公共类 SimpleAsyncTaskExecutor 扩展了 CustomizableThreadCreator 实现AsyncListenableTaskExecutor、Serializable

TaskExecutor 实现为每个任务启动一个新线程， 异步执行。

支持通过“concurrencyLimit”限制并发线程 豆属性。默认情况下，并发线程数为 无限制。

注意：此实现不重用线程！考虑一个 线程池 TaskExecutor 实现，特别是对于 执行大量短期任务。

Github 的使用示例：

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xmlns:beans="http://www.springframework.org/schema/beans"
             xsi:schemaLocation="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/integration
            http://www.springframework.org/schema/integration/spring-integration.xsd">

    <channel id="taskExecutorOnly">
        <dispatcher task-executor="taskExecutor"/>
    </channel>

    <channel id="failoverFalse">
        <dispatcher failover="false"/>
    </channel>

    <channel id="failoverTrue">
        <dispatcher failover="true"/>
    </channel>

    <channel id="loadBalancerDisabled">
        <dispatcher load-balancer="none"/>
    </channel>

    <channel id="loadBalancerDisabledAndTaskExecutor">
        <dispatcher load-balancer="none" task-executor="taskExecutor"/>
    </channel>

    <channel id="roundRobinLoadBalancerAndTaskExecutor">
        <dispatcher load-balancer="round-robin" task-executor="taskExecutor"/>
    </channel>

    <channel id="lbRefChannel">
        <dispatcher load-balancer-ref="lb"/>
    </channel>

    <beans:bean id="taskExecutor"
                class="org.springframework.core.task.SimpleAsyncTaskExecutor"/>

    <beans:bean id="lb"
                class="org.springframework.integration.channel.config.DispatchingChannelParserTests.SampleLoadBalancingStrategy"/>
</beans:beans>

Answer 2

从日志中我可以看到所有顺序但线程名称不同

因为日志只是打印消息的一个地方，而且它们实际上是由一个作者打印的，即使来自不同的线程。他们一一出现在那里。在良好的负载下，您肯定会看到消息以意外的顺序记录。

我假设 preSend/Postsend 应该以随机顺序调用。

那不是真的。拦截器的调用顺序是按照它们添加到通道的顺序以及它们的order 是否相同，这对你来说是一种情况。如何实现这些拦截器已经不是拦截器链的责任。

我认为您只是不幸运地看到以任意顺序排列的日志，可能只是因为这些执行程序通道的消费者是普通的记录器 - 没有任何负载来保持线程并且给人的印象是其他线程中的工作是并行完成的。