京公网安备11010502063487号Spring Integration 的 DirectChannel、QueueChannel、PublishSubscribeChannel 到底差在哪?消息发给谁、在哪个线程执行?处理器抛异常谁能看到?Gateway 的 replyTimeout 为什么没生效?poller 为什么会让消费变慢?这篇在 Spring Integration 7.0.6 上逐项实测。
发 4 条,处理情况(每条只给一个订阅者,轮流;在发送线程里执行):[h1[main]:m1, h2[main]:m2, h1[main]:m3, h2[main]:m4] 第一个订阅者总是抛异常,默认 failover:[h2:x1, h2:x2, h2:x3] setFailover(false):[异常:MessageDeliveryException, h2:x2, 异常:MessageDeliveryException]
点对点轮询,发送线程里执行,默认 failover。
没有 executor:两个订阅者都收到,都在发送线程里:[a[main], b[main]] 带 executor:[a[线程池线程], b[线程池线程]] 第一个订阅者抛异常(ignoreFailures 默认 false):[send 抛出 MessageDeliveryException]
广播;默认一个失败后面不执行。
容量 3 的队列已满,send(msg, 0) 返回 false send(msg, 500) 阻塞了 504 ms 后返回 false receive(0) 取到 q1 PollingConsumer(每 300 ms 轮询一次、每次最多取 1 条):6 条消息全部消费完用了 1566 ms maxMessagesPerPoll=-1(一次取完):6 条全部消费完用了 30 ms
poller 间隔与每次条数决定吞吐。
DirectChannel:send() 抛给了发送方:MessageDeliveryException → 原因 IllegalStateException ExecutorChannel:send() 返回 true,没有抛异常;订阅了 errorChannel 的处理器收到:[MessageDeliveryException←IllegalStateException]
同步通道抛给发送方,异步通道发到 errorChannel。
正常请求:call("hi") = reply:hi
处理器睡 1.5 秒、replyTimeout 500 ms:call("slow") = reply:slow,调用方等了 1505 ms
请求通道换成 ExecutorChannel(处理器在别的线程里跑):call("slow") = null,调用方等了 511 ms(replyTimeout 才起作用)
同一个 ExecutorChannel 网关,正常请求:call("hi") = reply:hi
处理器抛异常:调用方收到 IllegalStateException,原因 无同步通道上 replyTimeout 不生效。
输入 3 12 7 20 5:通过过滤(>5)的被转换后路由到 偶数=[12, 20] 奇数=[7];被过滤丢弃的=[3, 5]
DSL 把过滤、转换、路由写在一个 bean 里。