diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java index 384f3f12be..f8b74a02f9 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java @@ -93,7 +93,11 @@ public class ConsumeMessageConcurrentlyService implements ConsumeMessageService @Override public void run() { - cleanExpireMsg(); + try { + cleanExpireMsg(); + } catch (Throwable e) { + log.error("scheduleAtFixedRate cleanExpireMsg exception", e); + } } }, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES); diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java index 812e8ab87d..cc3aee4ef5 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java @@ -97,7 +97,11 @@ public class ConsumeMessageOrderlyService implements ConsumeMessageService { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { - ConsumeMessageOrderlyService.this.lockMQPeriodically(); + try { + ConsumeMessageOrderlyService.this.lockMQPeriodically(); + } catch (Throwable e) { + log.error("scheduleAtFixedRate lockMQPeriodically exception", e); + } } }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS); }