java - 使用 Spring 集成对消息进行速率限制

Question

我是 Spring 集成的新手。

我们有一个 REST 应用程序接收的消息数量(每分钟 6000 条消息)超出了数据库可以处理的数量。因此，我想将请求速率限制为每 15 秒 500 条消息(每分钟 2000 条)。我使用队列 channel 来实现这一点。

一段时间后，应用程序创建了 30,000 多个 Java 线程。此外，队列 channel 保存的消息多于队列容量中提到的消息。

如何减少线程数并限制Queue中的消息？

集成上下文 xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:int="http://www.springframework.org/schema/integration"
    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-5.0.xsd">

    <!-- Endpoint -->   
    <int:gateway service-interface="com.ratelimiter.PrintGateway" default-request-channel="inputChannel">
        <int:method name="print"/>  
    </int:gateway>

    <!-- Channel -->
    <int:channel id="inputChannel">
        <int:queue capacity="30000"/>
    </int:channel>

    <!-- Endpoint -->   
    <int:service-activator ref="receiver" input-channel="inputChannel" method="save">
        <int:poller fixed-rate="15" time-unit="SECONDS" max-messages-per-poll="500"></int:poller>
    </int:service-activator>

    <!--  Spring Bean -->
    <bean id="receiver" class="com.ratelimiter.saveToDataStore"/>

</beans>

PrintGateway接口(interface):

public interface PrintGateway {

    public Future<Message<String>> print(Message<?> message);
}

Answer 1

因为您的网关签名将返回 Future<Message<String>> ，这被视为异步网关:https://docs.spring.io/spring-integration/docs/5.0.6.RELEASE/reference/html/messaging-endpoints-chapter.html#async-gateway

默认情况下它使用

private volatile AsyncTaskExecutor asyncExecutor = new SimpleAsyncTaskExecutor();

这确实为每条新消息启动了一个新线程。重要的是:它等待回复来满足 Future 。根据您的代码，不会有任何回复，因此，网关中的线程会等待很长时间。

您应该考虑将网关签名更改为 void返回类型。这样您就可以真正发送并忘记。不会无缘无故地产生任何后台额外线程。