spring-boot - 添加 5th Rabbit Listener 导致 Quartz Job 卡在阻塞状态
问题描述
我有一个包含 5 个 RabbitListener 的 Spring Boot 应用程序。我还有 2 个定期运行的 Quartz 调度程序作业。当我启动我的 Spring Boot 应用程序时,这些作业似乎运行了一次,但随后保持在 BLOCKED 的触发状态并且不再运行。我正在使用 Spring Boot 2.1.2 和 Quartz 2.3.0。任何人都可以建议吗?
我曾经有 4 个 Rabbit Listeners 并没有遇到这个问题,但是添加第 5 个会导致这种情况发生。我已经尝试删除现有的 Rabbit 监听器,所以只有 4 个 Rabbit 监听器,包括我的新监听器,问题就消失了。
这是我要添加的兔子侦听器:
@Slf4j
@Component
public class MyListener
{
@Autowired
private MyProcess process;
@RabbitListener(queues = "#{T(com.earthport.elm.amqp.RabbitBinding).LIQUIDITY_BALANCE_CREDITED.getQueue()}",
containerFactory = "noDeadLetterRabbitListenerContainerFactory")
public void onMessage(MyMessage message)
{
if (log.isDebugEnabled())
{
log.debug("Received message: {}", message);
}
try
{
process.run(message);
}
catch (Exception e)
{
throw new AmqpRejectAndDontRequeueException(e);
}
}
}
这是我的侦听器容器工厂配置:
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
Jackson2JsonMessageConverter messageConverter,
RetryAdviceChainFactory deadLetterRetryAdviceChainFactory)
{
final SimpleRabbitListenerContainerFactory cf = new SimpleRabbitListenerContainerFactory();
cf.setConnectionFactory(connectionFactory);
cf.setConcurrentConsumers(1);
cf.setErrorHandler(defaultErrorHandler);
cf.setAdviceChain(deadLetterRetryAdviceChainFactory.createDefaultRetryChain());
cf.setChannelTransacted(true);
cf.setTaskExecutor(taskExecutor);
cf.setMessageConverter(messageConverter);
cf.setTransactionManager(transactionManager);
return cf;
}
@Bean
public SimpleRabbitListenerContainerFactory noDeadLetterRabbitListenerContainerFactory(
Jackson2JsonMessageConverter messageConverter,
RetryAdviceChainFactory loggingRetryAdviceChainFactory)
{
final SimpleRabbitListenerContainerFactory cf = new SimpleRabbitListenerContainerFactory();
cf.setConnectionFactory(connectionFactory);
cf.setConcurrentConsumers(1);
cf.setErrorHandler(defaultErrorHandler);
cf.setAdviceChain(loggingRetryAdviceChainFactory.createRetryChain(
new NoBackOffPolicy(), new NeverRetryPolicy()));
cf.setChannelTransacted(true);
cf.setTaskExecutor(taskExecutor);
cf.setMessageConverter(messageConverter);
cf.setTransactionManager(transactionManager);
return cf;
}
这是我的侦听器容器工厂配置:
@Bean
public JobFactory jobFactory(ApplicationContext applicationContext)
{
AutowiringSpringBeanJobFactory jobFactory = new AutowiringSpringBeanJobFactory();
jobFactory.setApplicationContext(applicationContext);
return jobFactory;
}
@Bean
public SchedulerFactoryBean schedulerFactoryBean(
DataSource dataSource,
JobFactory jobFactory,
Executor taskExecutor,
PlatformTransactionManager platformTransactionManager)
{
Properties quartzProperties = new Properties();
quartzProperties.setProperty("org.quartz.scheduler.instanceName", "ELM-Scheduler");
quartzProperties.setProperty("org.quartz.scheduler.instanceId", "AUTO");
quartzProperties.setProperty("org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX");
quartzProperties.setProperty("org.quartz.jobStore.driverDelegateClass", "org.quartz.impl.jdbcjobstore.StdJDBCDelegate");
quartzProperties.setProperty("org.quartz.jobStore.useProperties", "true");
quartzProperties.setProperty("org.quartz.jobStore.tablePrefix", "QRTZ_");
quartzProperties.setProperty("org.quartz.jobStore.isClustered", "true");
SchedulerFactoryBean factory = new SchedulerFactoryBean();
factory.setOverwriteExistingJobs(true);
factory.setAutoStartup(true);
factory.setDataSource(dataSource);
factory.setJobFactory(jobFactory);
factory.setTaskExecutor(taskExecutor);
factory.setTransactionManager(platformTransactionManager);
factory.setQuartzProperties(quartzProperties);
if (jobs != null)
{
factory.setJobDetails(jobs.toArray(new JobDetail[jobs.size()]));
}
if (triggers != null && schedulerEnabled)
{
factory.setTriggers(triggers.toArray(new Trigger[triggers.size()]));
}
return factory;
}
@Bean
public Scheduler scheduler(SchedulerFactoryBean schedulerFactoryBean)
{
return schedulerFactoryBean.getScheduler();
}
public static SimpleTriggerFactoryBean createTrigger(JobDetail jobDetail, long pollFrequencyMs, long startDelayMillis)
{
SimpleTriggerFactoryBean factoryBean = new SimpleTriggerFactoryBean();
factoryBean.setJobDetail(jobDetail);
factoryBean.setStartDelay(startDelayMillis);
factoryBean.setRepeatInterval(pollFrequencyMs);
factoryBean.setRepeatCount(SimpleTrigger.REPEAT_INDEFINITELY);
factoryBean.setMisfireInstruction(SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT);
return factoryBean;
}
这是我的 Quartz 和 RabbitConfig 中使用的共享 TaskExcutor:
@Configuration
@EnableAsync
public class AsyncConfig
{
@Value("${executor.maxPoolSize:50}")
private int maxPoolSize;
@Value("${executor.maxQueueCapcity:500}")
private int maxQueueCapacity;
@Bean
public Executor taskExecutor()
{
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(maxPoolSize);
executor.setQueueCapacity(maxQueueCapacity);
return executor;
}
}
我希望我的预定作业能够定期运行。当这工作时,我可以在我的 QRTZ_TRIGGERS 表中看到 TRIGGER_STATE 没有被阻塞。但是,添加我的第 5 个侦听器时,此值将变为 BLOCKED 并保留在那里。
解决方案
推荐阅读
- r - 在 Azure 机器学习工作室中提取线性回归的特征权重
- java - 如何反转具有换行符(\n 或 \r)的字符串中的单词?
- autolayout - UICollectionViewCell 内的 UIStackView 不在屏幕上
- mysql - mysql 按周分组并每周添加总计
- database - 如何删除 MongoDB 中的所有数据库?
- c# - 滚动 Bing Maps Control UWP C# 时更改光标样式
- c# - c# Regex 匹配所有斯里兰卡电话号码
- java - JSoup get text and inline images in order
- qt - 我如何告诉 Qt 它应该在哪里寻找 openssl?
- jquery - Choppy CSS 悬停动画有多个元素