Separate scheduler for subs (#603)

This commit is contained in:
KirillPamPam
2024-12-10 13:58:21 +04:00
committed by GitHub
parent dd3d521375
commit f138c035ed
2 changed files with 11 additions and 2 deletions

View File

@@ -25,6 +25,8 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
callTargetsHolder: CallTargetsHolder,
@Qualifier("headScheduler")
headScheduler: Scheduler,
@Qualifier("subScheduler")
subScheduler: Scheduler,
tracer: Tracer,
multistreamEventsScheduler: Scheduler,
indexConfig: IndexConfig,
@@ -41,6 +43,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
chain,
cachesFactory,
headScheduler,
subScheduler,
tracer,
multistreamEventsScheduler,
indexConfig.getByChain(chain),
@@ -54,6 +57,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
chain: Chain,
cachesFactory: CachesFactory,
headScheduler: Scheduler,
subScheduler: Scheduler,
tracer: Tracer,
multistreamEventsScheduler: Scheduler,
logsOracleConfig: IndexConfig.Index?,
@@ -71,7 +75,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
headScheduler,
cs.makeCachingReaderBuilder(tracer),
cs::localReaderBuilder,
cs.subscriptionBuilder(headScheduler),
cs.subscriptionBuilder(subScheduler),
logsOracleConfig,
logsOracleScheduler,
tracer,

View File

@@ -41,6 +41,11 @@ open class SchedulersConfig {
return makeScheduler("head-scheduler", 4, monitoringConfig)
}
@Bean
open fun subScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("sub-scheduler", 4, monitoringConfig)
}
@Bean
open fun logsOracleScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("logs-oracle", 4, monitoringConfig)
@@ -77,7 +82,7 @@ open class SchedulersConfig {
}
private fun makeScheduler(name: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler {
return Schedulers.fromExecutorService(makePool(name, size * threadsMultiplier, monitoringConfig))
return Schedulers.fromExecutorService(makePool(name, size, monitoringConfig))
}
private fun makePool(name: String, size: Int, monitoringConfig: MonitoringConfig): ExecutorService {