opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9898] Remove AbstractBrokerRunnable and replace with Runnable

Browse Source 
Change-Id: I94104151c452d09cbe195a3e1126a473b662a337

Co-authored-by: RongtongJin <user@example.com>
This commit is contained in:
rongtong
2025-12-05 14:03:54 +08:00
committed by GitHub
parent 1bb2168bf3
commit 89e021e871
9 changed files with 45 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+6 -7
View File
@@ -121,7 +121,6 @@ import org.apache.rocketmq.broker.transaction.queue.DefaultTransactionalMessageC
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageBridge;
import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageServiceImpl;
import org.apache.rocketmq.broker.util.HookUtils;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.BrokerIdentity;
import org.apache.rocketmq.common.MixAll;
@@ -1814,9 +1813,9 @@ public class BrokerController {
this.registerBrokerAll(true, false, true);
}
scheduledFutures.add(this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
scheduledFutures.add(this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
if (System.currentTimeMillis() < shouldStartTime) {
BrokerController.LOG.info("Register to namesrv after {}", shouldStartTime);
@@ -1836,9 +1835,9 @@ public class BrokerController {
if (this.brokerConfig.isEnableSlaveActingMaster()) {
scheduleSendHeartbeat();
scheduledFutures.add(this.syncBrokerMemberGroupExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
scheduledFutures.add(this.syncBrokerMemberGroupExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
BrokerController.this.syncBrokerMemberGroup();
} catch (Throwable e) {
@@ -1869,9 +1868,9 @@ public class BrokerController {
}
protected void scheduleSendHeartbeat() {
scheduledFutures.add(this.brokerHeartbeatExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
scheduledFutures.add(this.brokerHeartbeatExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
if (isIsolated) {
return;
}
broker/src/main/java/org/apache/rocketmq/broker/client/DefaultConsumerIdsChangeListener.java
+2 -3
View File
@@ -26,7 +26,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -48,9 +47,9 @@ public class DefaultConsumerIdsChangeListener implements ConsumerIdsChangeListen
public DefaultConsumerIdsChangeListener(BrokerController brokerController) {
this.brokerController = brokerController;
scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(brokerController.getBrokerConfig()) {
scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
notifyConsumerChange();
} catch (Exception e) {
broker/src/main/java/org/apache/rocketmq/broker/latency/BrokerFastFailure.java
+2 -3
View File
@@ -23,7 +23,6 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.UtilAll;
@@ -80,9 +79,9 @@ public class BrokerFastFailure {
}
public void start() {
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.brokerController.getBrokerConfig()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
if (brokerController.getBrokerConfig().isBrokerFastFailureEnable()) {
cleanExpiredRequest();
}
broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+15 -20
View File
@@ -46,7 +46,6 @@ import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.impl.consumer.PullResultExt;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.BrokerIdentity;
import org.apache.rocketmq.common.LockCallback;
import org.apache.rocketmq.common.MixAll;
@@ -356,18 +355,14 @@ public class BrokerOuterAPI {
requestHeader.setClusterName(clusterName);
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new AbstractBrokerRunnable(new BrokerIdentity(clusterName, brokerName, brokerId, isInBrokerContainer)) {
brokerOuterExecutor.execute(() -> {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_DATA_VERSION, requestHeader);
request.setBody(dataVersion.encode());
@Override
public void run0() {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_DATA_VERSION, requestHeader);
request.setBody(dataVersion.encode());
try {
BrokerOuterAPI.this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMillis);
} catch (Exception e) {
LOGGER.error("sendHeartbeat Exception " + namesrvAddr, e);
}
try {
BrokerOuterAPI.this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMillis);
} catch (Exception e) {
LOGGER.error("sendHeartbeat Exception " + namesrvAddr, e);
}
});
}
@@ -389,9 +384,9 @@ public class BrokerOuterAPI {
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new AbstractBrokerRunnable(new BrokerIdentity(clusterName, brokerName, brokerId, isInBrokerContainer)) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run0() {
public void run() {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.BROKER_HEARTBEAT, requestHeader);
try {
@@ -532,9 +527,9 @@ public class BrokerOuterAPI {
requestHeader.setBodyCrc32(bodyCrc32);
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new AbstractBrokerRunnable(brokerIdentity) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run0() {
public void run() {
try {
RegisterBrokerResult result = registerBroker(namesrvAddr, oneway, timeoutMills, requestHeader, body);
if (result != null) {
@@ -719,9 +714,9 @@ public class BrokerOuterAPI {
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new AbstractBrokerRunnable(new BrokerIdentity(clusterName, brokerName, brokerId, isInBrokerContainer)) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run0() {
public void run() {
try {
QueryDataVersionRequestHeader requestHeader = new QueryDataVersionRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
@@ -1501,9 +1496,9 @@ public class BrokerOuterAPI {
requestHeader.setHeartbeatTimeoutMills(controllerHeartBeatTimeoutMills);
requestHeader.setElectionPriority(electionPriority);
requestHeader.setBrokerId(brokerId);
brokerOuterExecutor.execute(new AbstractBrokerRunnable(new BrokerIdentity(clusterName, brokerName, brokerId, isInBrokerContainer)) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run0() {
public void run() {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.BROKER_HEARTBEAT, requestHeader);
try {
common/src/main/java/org/apache/rocketmq/common/AbstractBrokerRunnable.java
-48
View File
@@ -1,48 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.common;
import java.io.File;
import org.apache.rocketmq.logging.org.slf4j.MDC;
public abstract class AbstractBrokerRunnable implements Runnable {
protected final BrokerIdentity brokerIdentity;
public AbstractBrokerRunnable(BrokerIdentity brokerIdentity) {
this.brokerIdentity = brokerIdentity;
}
private static final String MDC_BROKER_CONTAINER_LOG_DIR = "brokerContainerLogDir";
/**
* real logic for running
*/
public abstract void run0();
@Override
public void run() {
try {
if (brokerIdentity.isInBrokerContainer()) {
MDC.put(MDC_BROKER_CONTAINER_LOG_DIR, File.separator + brokerIdentity.getCanonicalName());
}
run0();
} finally {
MDC.clear();
}
}
}
container/src/main/java/org/apache/rocketmq/container/BrokerContainer.java
+6 -7
View File
@@ -22,7 +22,6 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerPathConfigHelper;
import org.apache.rocketmq.broker.ConfigContext;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.BrokerIdentity;
import org.apache.rocketmq.common.MixAll;
@@ -156,9 +155,9 @@ public class BrokerContainer implements IBrokerContainer {
this.updateNamesrvAddr();
LOG.info("Set user specified name server address: {}", this.brokerContainerConfig.getNamesrvAddr());
// also auto update namesrv if specify
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(BrokerIdentity.BROKER_CONTAINER_IDENTITY) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
BrokerContainer.this.updateNamesrvAddr();
} catch (Throwable e) {
@@ -167,10 +166,10 @@ public class BrokerContainer implements IBrokerContainer {
}
}, 1000 * 10, this.brokerContainerConfig.getUpdateNamesrvAddrInterval(), TimeUnit.MILLISECONDS);
} else if (this.brokerContainerConfig.isFetchNamesrvAddrByAddressServer()) {
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(BrokerIdentity.BROKER_CONTAINER_IDENTITY) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
BrokerContainer.this.brokerOuterAPI.fetchNameServerAddr();
} catch (Throwable e) {
@@ -180,9 +179,9 @@ public class BrokerContainer implements IBrokerContainer {
}, 1000 * 10, this.brokerContainerConfig.getFetchNamesrvAddrInterval(), TimeUnit.MILLISECONDS);
}
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(BrokerIdentity.BROKER_CONTAINER_IDENTITY) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
BrokerContainer.this.brokerOuterAPI.refreshMetadata();
} catch (Exception e) {
container/src/main/java/org/apache/rocketmq/container/InnerBrokerController.java
+4 -5
View File
@@ -21,7 +21,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.auth.config.AuthConfig;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.RemotingServer;
@@ -82,9 +81,9 @@ public class InnerBrokerController extends BrokerController {
this.registerBrokerAll(true, false, true);
}
scheduledFutures.add(this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
scheduledFutures.add(this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
if (System.currentTimeMillis() < shouldStartTime) {
BrokerController.LOG.info("Register to namesrv after {}", shouldStartTime);
@@ -104,9 +103,9 @@ public class InnerBrokerController extends BrokerController {
if (this.brokerConfig.isEnableSlaveActingMaster()) {
scheduleSendHeartbeat();
scheduledFutures.add(this.syncBrokerMemberGroupExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
scheduledFutures.add(this.syncBrokerMemberGroupExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
try {
InnerBrokerController.this.syncBrokerMemberGroup();
} catch (Throwable e) {
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+8 -9
View File
@@ -57,7 +57,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.BrokerIdentity;
@@ -1775,23 +1774,23 @@ public class DefaultMessageStore implements MessageStore {
private void addScheduleTask() {
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
DefaultMessageStore.this.cleanFilesPeriodically();
}
}, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
DefaultMessageStore.this.checkSelf();
}
}, 1, 10, TimeUnit.MINUTES);
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
if (DefaultMessageStore.this.getMessageStoreConfig().isDebugLockEnable()) {
try {
if (DefaultMessageStore.this.commitLog.getBeginTimeInLock() != 0) {
@@ -1810,9 +1809,9 @@ public class DefaultMessageStore implements MessageStore {
}
}, 1, 1, TimeUnit.SECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run0() {
public void run() {
DefaultMessageStore.this.storeCheckpoint.flush();
}
}, 1, 1, TimeUnit.SECONDS);
store/src/main/java/org/apache/rocketmq/store/index/IndexService.java
+2 -3
View File
@@ -23,7 +23,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.rocketmq.common.AbstractBrokerRunnable;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -339,9 +338,9 @@ public class IndexService {
if (indexFile != null) {
final IndexFile flushThisFile = prevIndexFile;
Thread flushThread = new Thread(new AbstractBrokerRunnable(defaultMessageStore.getBrokerConfig()) {
Thread flushThread = new Thread(new Runnable() {
@Override
public void run0() {
public void run() {
IndexService.this.flush(flushThisFile);
}
}, "FlushIndexFileThread");