Spring Integration 轻松实现服务间消息传递，真香！

昨天晚上十二点多吧，我在公司楼下喝奶茶…不是，咖啡，奶茶太腻…结果手机突然震一下，群里有人问我：“东哥你之前不是说过 Spring Integration 很轻松能搞微服务之间传消息么？到底咋轻松？” 我当时哈，脑子已经半糊了，但还是忍不住嘴欠讲了半小时…现在说给你们听听…

我先喝口水…嗯继续…

你们知道吧，之前我们组那个小李，他就是写服务之间调用，每次都用 Feign，啥都 Feign，我说兄弟你不能啥都同步请求，对吧，你得异步啊。结果他死活嫌麻烦。后来我们搞一个库存系统和结算系统的对接，他一个同步链条啊…我靠，那延迟飙到两百多毫秒，像踩油门的三蹦子似的。

我那天就跟他说，“你用 Spring Integration 啊老弟，它这个…呃说简单点就是把消息在线程之间、服务之间…传一传，像你扔纸条一样，你不用管对面啥时候看，只管往管道里丢。” 他当时还说“难吧？”我直接让他看了示例代码，他当场闭麦。

我找下那个代码…等下…

@Bean
public MessageChannel orderChannel() {
    return new DirectChannel();  // 这个是直接通道，丢进去就走
}

@Bean
public IntegrationFlow orderFlow() {
    return IntegrationFlows.from("orderChannel")
            .handle(msg -> {
                System.out.println("收到订单消息: " + msg.getPayload());
            })
            .get();
}

你看，就这样…我靠是不是很简单… 你只要 orderChannel().send(new GenericMessage<>("hello"))，它就啪一下跑到那个 handle 里了，你连 MQ 都不用连。 小李当时说：“东哥这不就是个本地异步？” 我说：“哎你别急，你把 Binder 换成 Kafka、RabbitMQ，那就是跨服务了兄弟。”

然后那天中午吃饭的时候，诶对了，我们那边食堂炸鸡腿真的巨干，你们千万别去…吃着吃着我跟他说 Spring Integration 的那个网关，他突然就开窍了。就是那个 Gateway，它能像方法调用一样，把消息丢到 Integration Flow 里，你代码看着还像同步，实际上里面是消息驱动。

@MessagingGateway(defaultRequestChannel = "payChannel")
public interface PayGateway {
    String callPay(String payload);
}

当时他愣住了说：“还能这样写？这不是比 Feign 还丝滑？” 我说是啊，但你别乱用，你要是把每条链路都改成异步你后面排查问题能找到你祖宗十八代。

我突然想起来我手机忘充电了…算了继续说…

你们记得我们有个业务是，每来一笔订单都要发短信、同时记账、同时写日志吧？以前我们三个线程池搞得跟菜市场似的，线程到处乱跑。 后来我用 Spring Integration 的 Publish-Subscribe flow，就是那个一进多出的。

就是这样——我打给你们看：

@Bean
public IntegrationFlow orderPubFlow() {
    return IntegrationFlows.from("orderIn")
            .publishSubscribeChannel(s -> s
                    .subscribe(f -> f.handle(msg -> sms(msg)))
                    .subscribe(f -> f.handle(msg -> billing(msg)))
                    .subscribe(f -> f.handle(msg -> audit(msg)))
            ).get();
}

一条消息，啪一下三份，省事得很。 我们那天在会议室调试，搞到十一点多，运维都要锁门了还催我们走…

不过这个东西我跟你们说噢，它好用是好用，但是你别…别以为它跟 Kafka 那种分布式消息一样能抗大流量。Spring Integration 更像是把你本地的消息流理顺，然后再配合 Spring Cloud Stream 去接 MQ。 你要拿它硬抗 10 万 TPS 我只能说你人没了。

哦对对我突然想到一个小坑，小李踩过： 他之前定义了一个 QueueChannel，然后疯狂丢消息进去不消费，结果内存直接爆了。 QueueChannel 默认是无界的，他不知道，你们千万别学他。

@Bean
public MessageChannel riskQueue() {
    return new QueueChannel(2000); // 指定容量！！！
}

一定要设容量，不然你排查 OOM 排到怀疑人生… 那天凌晨两点他电话打我，我还以为地震了。

啥…我刚讲到哪了…等一下我想想… 哦对，“轻松实现消息传递真香”这个事—— 其实真香的点不在于“会不会用”，而是你用了之后，你发现你原来代码里那种牵线木偶一样的依赖关系全没了。

有时候你一个请求到了服务 A，它丢给 Integration Flow，Flow 里异步跑三四个步骤，然后结果再给你返回…你压根不知道它里面搞了多大动静，代码却看着很干净。 就这点，小李天天说“东哥我悟了悟了”，我说悟你妹你把注释写清楚先。

哎我现在想喝冰美式…算了不说了我下楼买杯咖啡先，你要我继续讲怎么跟 MQ 接？还是讲它和 Spring Cloud Stream 的区别？你发我我回来继续说。