8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Broker增加服务是否存活接口(#654)

Browse Source 
Broker增加服务是否存活接口
This commit is contained in:
EricZeng
2022-10-10 19:56:12 +08:00
committed by GitHub
parent 81b1ec48c2 87cd058fd8
commit 9cf13e9b30
2 changed files with 24 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/BrokerService.java
View File

@@ -67,4 +67,8 @@ public interface BrokerService {
* 获取总的Broker数
*/
Integer countAllBrokers();
boolean allServerDown(Long clusterPhyId);
boolean existServerDown(Long clusterPhyId);
}

22
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/broker/impl/BrokerServiceImpl.java
View File

@@ -262,14 +262,32 @@ public class BrokerServiceImpl extends BaseVersionControlService implements Brok
return version;
}
@Override
public Integer countAllBrokers() {
LambdaQueryWrapper<BrokerPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
return brokerDAO.selectCount(lambdaQueryWrapper);
}
@Override
public boolean allServerDown(Long clusterPhyId) {
List<BrokerPO> poList = this.getAllBrokerPOsFromDB(clusterPhyId);
if (ValidateUtils.isEmptyList(poList)) {
return false;
}
return poList.stream().filter(elem -> elem.getStatus().equals(Constant.DOWN)).count() == poList.size();
}
@Override
public boolean existServerDown(Long clusterPhyId) {
List<BrokerPO> poList = this.getAllBrokerPOsFromDB(clusterPhyId);
if (ValidateUtils.isEmptyList(poList)) {
return false;
}
return poList.stream().filter(elem -> elem.getStatus().equals(Constant.DOWN)).count() > 0;
}
/**************************************************** private method ****************************************************/
private List<Broker> listAllBrokersAndUpdateCache(Long clusterPhyId) {