opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 05:46:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #6910] Extract the interval of fetchNameServerAddr

Browse Source 
Co-authored-by: RongtongJin <jinrongtong16@mails.ucas.ac.cn>
This commit is contained in:
Guocheng Tang
2023-06-23 14:21:29 +08:00
committed by GitHub
parent fd4a397a14
commit 5d6ee6384f
5 changed files with 71 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+27 -26
View File
@@ -17,31 +17,6 @@
package org.apache.rocketmq.broker;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.acl.plain.PlainAccessValidator;
import org.apache.rocketmq.broker.client.ClientHousekeepingService;
@@ -159,6 +134,32 @@ import org.apache.rocketmq.store.timer.TimerCheckpoint;
import org.apache.rocketmq.store.timer.TimerMessageStore;
import org.apache.rocketmq.store.timer.TimerMetrics;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;
public class BrokerController {
protected static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private static final Logger LOG_PROTECTION = LoggerFactory.getLogger(LoggerName.PROTECTION_LOGGER_NAME);
@@ -720,7 +721,7 @@ public class BrokerController {
LOG.error("Failed to fetch nameServer address", e);
}
}
}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
}, 1000 * 10, this.brokerConfig.getFetchNamesrvAddrInterval(), TimeUnit.MILLISECONDS);
}
}
common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+15 -1
View File
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.common;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.annotation.ImportantField;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.MessageRequestMode;
@@ -24,6 +23,8 @@ import org.apache.rocketmq.common.metrics.MetricsExporterType;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.NetworkUtil;
import java.util.concurrent.TimeUnit;
public class BrokerConfig extends BrokerIdentity {
private String brokerConfigPath = null;
@@ -374,6 +375,11 @@ public class BrokerConfig extends BrokerIdentity {
private boolean usePIDColdCtrStrategy = true;
private long cgColdReadThreshold = 3 * 1024 * 1024;
private long globalColdReadThreshold = 100 * 1024 * 1024;
/**
* The interval to fetch namesrv addr, default value is 10 second
*/
private long fetchNamesrvAddrInterval = 10 * 1000;
public long getMaxPopPollingSize() {
return maxPopPollingSize;
@@ -1662,4 +1668,12 @@ public class BrokerConfig extends BrokerIdentity {
public void setUseStaticSubscription(boolean useStaticSubscription) {
this.useStaticSubscription = useStaticSubscription;
}
public long getFetchNamesrvAddrInterval() {
return fetchNamesrvAddrInterval;
}
public void setFetchNamesrvAddrInterval(final long fetchNamesrvAddrInterval) {
this.fetchNamesrvAddrInterval = fetchNamesrvAddrInterval;
}
}
common/src/main/java/org/apache/rocketmq/common/BrokerIdentity.java
+3 -2
View File
@@ -17,8 +17,6 @@
package org.apache.rocketmq.common;
import java.net.InetAddress;
import java.net.UnknownHostException;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
@@ -27,6 +25,9 @@ import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
public class BrokerIdentity {
private static final String DEFAULT_CLUSTER_NAME = "DefaultCluster";
container/src/main/java/org/apache/rocketmq/container/BrokerContainer.java
+13 -12
View File
@@ -16,17 +16,6 @@
*/
package org.apache.rocketmq.container;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerPathConfigHelper;
@@ -50,6 +39,18 @@ import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class BrokerContainer implements IBrokerContainer {
private static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -177,7 +178,7 @@ public class BrokerContainer implements IBrokerContainer {
LOG.error("ScheduledTask fetchNameServerAddr exception", e);
}
}
}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
}, 1000 * 10, this.brokerContainerConfig.getFetchNamesrvAddrInterval(), TimeUnit.MILLISECONDS);
}
this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(BrokerIdentity.BROKER_CONTAINER_IDENTITY) {
container/src/main/java/org/apache/rocketmq/container/BrokerContainerConfig.java
+13 -1
View File
@@ -38,6 +38,11 @@ public class BrokerContainerConfig {
private String brokerContainerIP = NetworkUtil.getLocalAddress();
private String brokerConfigPaths = null;
/**
* The interval to fetch namesrv addr, default value is 10 second
*/
private long fetchNamesrvAddrInterval = 10 * 1000;
public String getRocketmqHome() {
return rocketmqHome;
@@ -82,5 +87,12 @@ public class BrokerContainerConfig {
public void setBrokerConfigPaths(String brokerConfigPaths) {
this.brokerConfigPaths = brokerConfigPaths;
}
public long getFetchNamesrvAddrInterval() {
return fetchNamesrvAddrInterval;
}
public void setFetchNamesrvAddrInterval(final long fetchNamesrvAddrInterval) {
this.fetchNamesrvAddrInterval = fetchNamesrvAddrInterval;
}
}