我使用quarkus smallrye与kafka的React式消息传递获得以下stacktrace：
2020-07-24 01:38:31,662 ERROR [io.sma.rea.mes.kafka] (executor-thread-870) SRMSG18207: Unable to dispatch message to Kafka: io.smallrye.mutiny.subscription.BackPressureFailure: Could not emit tick 211 due to lack of requests
at io.smallrye.mutiny.operators.multi.builders.IntervalMulti$IntervalRunnable.run(IntervalMulti.java:83)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at org.jboss.threads.ContextClassLoaderSavingRunnable.run(ContextClassLoaderSavingRunnable.java:35)
at org.jboss.threads.EnhancedQueueExecutor.safeRun(EnhancedQueueExecutor.java:2046)
at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.doRunTask(EnhancedQueueExecutor.java:1578)
at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1452)
at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)
at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)
at java.lang.Thread.run(Thread.java:748)
at org.jboss.threads.JBossThread.run(JBossThread.java:479)所以我读过https://smallrye.io/smallrye-mutiny/#_how_do_i_control_the_back_pressure
根据文件我已经添加了背压控制。
之前：
@Outgoing( "eqs-crossing-xxx" )
public Multi< EQSAlert > eqsCrossingXXX_XXX(){
final String series = CrossingEnum.XXX_XXX.getSeries();
final String equipment = CrossingEnum.XXX_XXX.getEquipment();
final String vehicleRegex = vehicleRegexService.getRegexBySeries( series );
log.info( "Incoming request for {} - {}", series, equipment);
log.info( "Vehicle regex : {}", vehicleRegex );
return Multi
.createFrom()
.ticks()
.every(
Duration.ofSeconds( poolingInterval )
)
.concatMap(i -> {
final Multi<CrossingState> crossingStateBySeriesAndEquipment = CrossingState.getCrossingStateBySeriesAndEquipment(client, series, equipment);
return crossingStateBySeriesAndEquipment.flatMap(crossingState ->
crossingState.isActive() ?
EQSAlert.getEQSAlertBySeriesAndEquipment(
client,
series,
vehicleRegex,
equipment
)
:
Multi.createFrom().empty()
);
});
}之后：
@Outgoing( "eqs-crossing-xxx" )
public Multi< EQSAlert > eqsCrossingXXX_XXX(){
final String series = CrossingEnum.XXX_XXX.getSeries();
final String equipment = CrossingEnum.XXX_XXX.getEquipment();
final String vehicleRegex = vehicleRegexService.getRegexBySeries( series );
log.info( "Incoming request for {} - {}", series, equipment);
log.info( "Vehicle regex : {}", vehicleRegex );
return Multi
.createFrom()
.ticks()
.every(
Duration.ofSeconds( poolingInterval )
)
.onOverflow()
.buffer(10)
.concatMap(i -> {
final Multi<CrossingState> crossingStateBySeriesAndEquipment = CrossingState.getCrossingStateBySeriesAndEquipment(client, series, equipment);
return crossingStateBySeriesAndEquipment.flatMap(crossingState ->
crossingState.isActive() ?
EQSAlert.getEQSAlertBySeriesAndEquipment(
client,
series,
vehicleRegex,
equipment
)
:
Multi.createFrom().empty()
);
});
}现在一切都好了。
我这篇文章的目的是了解我为什么要这么做？
https://smallrye.io/smallrye-mutiny/apidocs/io/smallrye/mutiny/subscription/backpressurestrategy.html
为什么缓冲器不能留住你？
如您所见，我每5秒调用一个简单的sql函数（pooliginterval）。函数返回一些记录（池中的记录不超过10条）
所以车流量很低
我可以用一些词来理解缓冲区管理吗。
谢谢您
1条答案
按热度按时间7fyelxc51#
你的下游每单位时间只能消耗一定数量的物品。这取决于你在做什么。默认情况下，kafka将自身限制为5次并发写入（您可以对其进行配置）。
所以，如果你以一种不受限制的，而不是背压感知的方式排放更多的气体，那么下游就跟不上了。添加一个包含10个项目的缓冲区可能会处理一些小问题，但这可能还不够。
drop更激进；如果下游跟不上，这些项目就会被丢弃。dropPreviousItems正在删除已接收的项目。