From c5db5ac3e41bd9b922e507fba438dd89e223b2d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E6=99=93=E5=8F=8C=20Li=20Xiao=20Shuang?= <644968328@qq.com> Date: Wed, 6 Apr 2022 22:02:00 +0800 Subject: [PATCH] [ISSUE #4125] Anonymous new Runnable() can be replaced with lambda (#4031) --- .../rocketmq/broker/BrokerController.java | 124 +++++++----------- 1 file changed, 49 insertions(+), 75 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java index 7bfc618b92..772eec6dc4 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java @@ -202,15 +202,15 @@ public class BrokerController { this.slaveSynchronize = new SlaveSynchronize(this); - this.sendThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getSendThreadPoolQueueCapacity()); - this.putThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getPutThreadPoolQueueCapacity()); - this.pullThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getPullThreadPoolQueueCapacity()); - this.replyThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getReplyThreadPoolQueueCapacity()); - this.queryThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getQueryThreadPoolQueueCapacity()); - this.clientManagerThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getClientManagerThreadPoolQueueCapacity()); - this.consumerManagerThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getConsumerManagerThreadPoolQueueCapacity()); - this.heartbeatThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity()); - this.endTransactionThreadPoolQueue = new LinkedBlockingQueue(this.brokerConfig.getEndTransactionPoolQueueCapacity()); + this.sendThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getSendThreadPoolQueueCapacity()); + this.putThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getPutThreadPoolQueueCapacity()); + this.pullThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getPullThreadPoolQueueCapacity()); + this.replyThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getReplyThreadPoolQueueCapacity()); + this.queryThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getQueryThreadPoolQueueCapacity()); + this.clientManagerThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getClientManagerThreadPoolQueueCapacity()); + this.consumerManagerThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getConsumerManagerThreadPoolQueueCapacity()); + this.heartbeatThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity()); + this.endTransactionThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getEndTransactionPoolQueueCapacity()); this.brokerStatsManager = messageStoreConfig.isEnableLmq() ? new LmqBrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat()) : new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat()); @@ -350,70 +350,51 @@ public class BrokerController { final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis(); final long period = 1000 * 60 * 60 * 24; - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - BrokerController.this.getBrokerStats().record(); - } catch (Throwable e) { - log.error("schedule record error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.getBrokerStats().record(); + } catch (Throwable e) { + log.error("schedule record error.", e); } }, initialDelay, period, TimeUnit.MILLISECONDS); - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - BrokerController.this.consumerOffsetManager.persist(); - } catch (Throwable e) { - log.error("schedule persist consumerOffset error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.consumerOffsetManager.persist(); + } catch (Throwable e) { + log.error("schedule persist consumerOffset error.", e); } }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS); - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - BrokerController.this.consumerFilterManager.persist(); - } catch (Throwable e) { - log.error("schedule persist consumer filter error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.consumerFilterManager.persist(); + } catch (Throwable e) { + log.error("schedule persist consumer filter error.", e); } }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS); - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - BrokerController.this.protectBroker(); - } catch (Throwable e) { - log.error("protectBroker error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.protectBroker(); + } catch (Throwable e) { + log.error("protectBroker error.", e); } }, 3, 3, TimeUnit.MINUTES); - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - BrokerController.this.printWaterMark(); - } catch (Throwable e) { - log.error("printWaterMark error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.printWaterMark(); + } catch (Throwable e) { + log.error("printWaterMark error.", e); } }, 10, 1, TimeUnit.SECONDS); - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - - @Override - public void run() { - try { - log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes()); - } catch (Throwable e) { - log.error("schedule dispatchBehindBytes error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes()); + } catch (Throwable e) { + log.error("schedule dispatchBehindBytes error.", e); } }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS); @@ -421,15 +402,11 @@ public class BrokerController { this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr()); log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr()); } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) { - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - - @Override - public void run() { - try { - BrokerController.this.brokerOuterAPI.fetchNameServerAddr(); - } catch (Throwable e) { - log.error("ScheduledTask fetchNameServerAddr exception", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.brokerOuterAPI.fetchNameServerAddr(); + } catch (Throwable e) { + log.error("ScheduledTask fetchNameServerAddr exception", e); } }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS); } @@ -443,14 +420,11 @@ public class BrokerController { this.updateMasterHAServerAddrPeriodically = true; } } else { - this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - try { - BrokerController.this.printMasterAndSlaveDiff(); - } catch (Throwable e) { - log.error("schedule printMasterAndSlaveDiff error.", e); - } + this.scheduledExecutorService.scheduleAtFixedRate(() -> { + try { + BrokerController.this.printMasterAndSlaveDiff(); + } catch (Throwable e) { + log.error("schedule printMasterAndSlaveDiff error.", e); } }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS); }