提交 04c8925d 编写于 作者: J Jaskey 提交者: dongeforever

[ROCKETMQ-160]SendHeartBeat may not be logged in the expected period closes...

[ROCKETMQ-160]SendHeartBeat may not be logged in the expected period closes apache/incubator-rocketmq#86
上级 adae1624
...@@ -112,7 +112,7 @@ public class MQClientInstance { ...@@ -112,7 +112,7 @@ public class MQClientInstance {
private final RebalanceService rebalanceService; private final RebalanceService rebalanceService;
private final DefaultMQProducer defaultMQProducer; private final DefaultMQProducer defaultMQProducer;
private final ConsumerStatsManager consumerStatsManager; private final ConsumerStatsManager consumerStatsManager;
private final AtomicLong storeTimesTotal = new AtomicLong(0); private final AtomicLong sendHeartbeatTimesTotal = new AtomicLong(0);
private ServiceState serviceState = ServiceState.CREATE_JUST; private ServiceState serviceState = ServiceState.CREATE_JUST;
private DatagramSocket datagramSocket; private DatagramSocket datagramSocket;
private Random random = new Random(); private Random random = new Random();
...@@ -517,38 +517,40 @@ public class MQClientInstance { ...@@ -517,38 +517,40 @@ public class MQClientInstance {
return; return;
} }
long times = this.storeTimesTotal.getAndIncrement(); if (!this.brokerAddrTable.isEmpty()) {
Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator(); long times = this.sendHeartbeatTimesTotal.getAndIncrement();
while (it.hasNext()) { Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
Entry<String, HashMap<Long, String>> entry = it.next(); while (it.hasNext()) {
String brokerName = entry.getKey(); Entry<String, HashMap<Long, String>> entry = it.next();
HashMap<Long, String> oneTable = entry.getValue(); String brokerName = entry.getKey();
if (oneTable != null) { HashMap<Long, String> oneTable = entry.getValue();
for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) { if (oneTable != null) {
Long id = entry1.getKey(); for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) {
String addr = entry1.getValue(); Long id = entry1.getKey();
if (addr != null) { String addr = entry1.getValue();
if (consumerEmpty) { if (addr != null) {
if (id != MixAll.MASTER_ID) if (consumerEmpty) {
continue; if (id != MixAll.MASTER_ID)
} continue;
try {
int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);
if (!this.brokerVersionTable.containsKey(brokerName)) {
this.brokerVersionTable.put(brokerName, new HashMap<String, Integer>(4));
}
this.brokerVersionTable.get(brokerName).put(addr, version);
if (times % 20 == 0) {
log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
log.info(heartbeatData.toString());
} }
} catch (Exception e) {
if (this.isBrokerInNameServer(addr)) { try {
log.error("send heart beat to broker exception", e); int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);
} else { if (!this.brokerVersionTable.containsKey(brokerName)) {
log.info("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName, this.brokerVersionTable.put(brokerName, new HashMap<String, Integer>(4));
id, addr); }
this.brokerVersionTable.get(brokerName).put(addr, version);
if (times % 20 == 0) {
log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
log.info(heartbeatData.toString());
}
} catch (Exception e) {
if (this.isBrokerInNameServer(addr)) {
log.info("send heart beat to broker[{} {} {}] failed", brokerName, id, addr);
} else {
log.info("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName,
id, addr);
}
} }
} }
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册