标签：机器 String 自定义 shard private kafka loghub return public
一般地，像kafka之类的消息中间件，作为一个可以保持历史消息的组件，其消费模型一般是主动拉取方式。这是为了给消费者足够的自由，回滚或者前进。
然而，也正是由于将消费消息的权力交给了消费者，所以，消费者往往需要承担更多的责任。比如：需要自行保存消费偏移量，以便后续可以知道从哪里继续。而当这一点处理不好时，则可能带来一些麻烦。
不管怎么样，解决方案也都是现成的，咱们也不用担心。
今天我们要谈论的是一个场景： 如何让n个机器消费m个分片数据？
这在消息中间件的解决方案里，明白地写着，使用消费者群组就可以实现了。具体来说就是，每个分片至多会被一机器消费，每个机器则可以消费多个分片数据。即机器数据小于分片数时，分片会被均衡地分配到消费者中。当机器数大于分片数时，多余的机器将不做任何事情。
好吧，既然官方已经说明白了，那咱们应该就不再需要自己搞一个轮子了吧。
但是，我还有个场景：如果我要求在机器做负载重平衡时，需要保证被抽取出去的机器分片，至少保留一段时间，不允许任何机器消费该分片，因为可能还有数据需要备份。
针对这种场景，我想官方也许是有提供回调函数之类的解决方案的吧。不管了，反正我没找到，只能自己先造个轮子了。
本文场景前提:
1. 使用loghub作为消息中间件(原理同kafka);
2. 整个数据有m个分片shard;
3. 整个消费者集群有n台机器;
4. 每个分片的数据需要集中到一机器上做有状态处理;
5. 可以借助redis保存有状态数据,以便消费者机器做优雅停机;
最简单的方案是，使 n=m, 每台机器消费一个shard, 这样状态永远不会错乱。
但是这样明显可扩展能力太差了！
比如有时数据量小了，虽然分片还在，但是完全不用那么多机器的时候，如何缩减机器？
比如由于数据压力大了，我想增加下分片数，以提高发送者性能，但是消费者我还不想理他，消费慢点无所谓？
其实，我们可以使用官方的消费者群组方法，可以动态缩减机器。
但是这个有状态就比较难做到了。
以上痛点，总结下来就是，可扩展性问题。
想象中的轮子是怎么样的？
1. 需要有个注册中心，管理机器的上下线监控;
2. 需要有负载均衡器，负载将shard的负载均衡的分布到在线机器中;
3. 需要有每个机器自己消费的分片记录，以使机器自身有据可查;
4. 需要有每个分片的消费情况，以判定出哪些分片已分配给哪些机器;
我们来细看下实现:
【1】平衡协调器主框架:
import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.Shard;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.response.ListShardResponse;
import com.test.common.config.LogHubProperties;
import com.test.utils.RedisPoolUtil;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static com.test.dispatcher.work.RedisKeyConstants.MAX_CONSUMER_SHARD_LOAD;
/**
* loghub动态消费者 shard分配shard 协调器
*
*/
public class LoghubConsumerShardCoWorker implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(LoghubConsumerShardCoWorker.class);
private LogHubProperties logHubProperties;
private RedisPoolUtil redisPoolUtil;
private Client mClient;
private ShardAssignMaster shardAssignMaster;
private String HOST_NAME;
public LoghubConsumerShardCoWorker(RedisPoolUtil redisPoolUtil, LogHubProperties logHubProperties) {
this(redisPoolUtil, logHubProperties, null);
}
public LoghubConsumerShardCoWorker(RedisPoolUtil redisPoolUtil, LogHubProperties logHubProperties, String hostName) {
this.redisPoolUtil = redisPoolUtil;
this.logHubProperties = logHubProperties;
this.HOST_NAME = hostName;
initSharedVars();
initConsumerClient();
initShardAssigner();
getAllShardList();
registerSelfConsumer();
startHeartBeatThread();
}
/**
* 开启心跳线程，保活
*/
private void startHeartBeatThread() {
ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
executorService.scheduleAtFixedRate(() -> {
String serverConsumeCacheKey = RedisKeyConstants.SERVER_CONSUME_CACHE_KEY_PREFIX + HOST_NAME;
redisPoolUtil.expire(serverConsumeCacheKey, 30);
shardAssignMaster.sendHeartbeat(HOST_NAME);
}, 30, 25, TimeUnit.SECONDS);
}
/**
* 初始化客户端实例
*/
private void initConsumerClient() {
this.mClient = new Client(logHubProperties.getEndpoint(),
logHubProperties.getAccessKeyId(), logHubProperties.getAccessKey());
}
/**
* 初始化分片分配控制器
*/
private void initShardAssigner() {
shardAssignMaster = new ShardAssignMaster(redisPoolUtil);
}
/**
* 初始化公共变量
*/
private void initSharedVars() {
try {
if(HOST_NAME != null) {
return;
}
HOST_NAME = InetAddress.getLocalHost().getHostName();
}
catch (UnknownHostException e) {
logger.error("init error : 获取服务器主机名失败", e);
throw new RuntimeException("init error : 获取服务器主机名失败");
}
}
/**
* 将自己作为消费者注册到消费者列表中，以判定后续可以进行消费
*/
private void registerSelfConsumer() {
shardAssignMaster.registerConsumer(HOST_NAME);
shardAssignMaster.sendHeartbeat(HOST_NAME);
}
@Override
public void run() {
try {
checkConsumerSharding();
}
catch (Exception e) {
logger.error("动态分配shard 发生异常", e);
}
}
/**
* job 只做一件事，即检查 shard 的消费情况，不平衡则处理
*/
private void checkConsumerSharding() {
try {
if (tryCoWorkerLock()) {
// step1. 检查是否需要进行shard分配
// 集群消费loghub数据动态伸缩策略
// 1. 启动时先去获取部分片数，备用；
// 2. 应用启动后，把自己注册到注册中心或redis中；
// 3. 根据注册上来的机器列表，按平均分配策略分配shard（只能由一个机器来分配，其他机器处理分布式锁竞争失败，等待状态）；
// 4. 分配好后，释放锁，各机器开始消费，如机器A消费shard 0/3，则机器1以轮询的方式依次从shard 0/3 摘取数据消费；
// 5. 分配好的数据结构为：prefix+ip保存具体数据，另外将自己的key添加到另一个zset中，标识自己存活；自己的key有效期为30秒；使用另一维度 shard，保存每个shard被占用情况，使用hash保存，key为shard，value为当有占用时为机器ip或主机名，当无占用时为null或空串；
// 6. 以上数据刷入，将在机器抢占到shard更新数据；shard总数信息暂时不允许在运行期间进行变更；（即如果变理shard必须重启服务器）
// 7. 机器下线时，占用的key将自动过期；（考虑是否主动删除）
// 8. 各机器上启动一个后台扫描线程，每隔30秒扫描一次。扫描zset，取出所有值后查看是否存在相应的key，如果不存在说明机器已下线，需要重新分配其占用的shard；
// 9. 重新分配策略，使用一致性hash算法实现；
// 10. 机器上线时，使用一致性hash算法重新平衡shard；
// 11. 使用分布式锁保证分配进程只有一个；
CheckConsumerShardingResultContainer resultContainer = checkShardConsumerReBalanceStatus();
if(resultContainer.getStatusResultType() != ReBalanceStatusResultEnum.OK) {
reBalanceConsumerShard(resultContainer);
}
}
}
finally {
releaseCoWorkerLock();
}
}
/**
* 确认机器和shard是否需要再平衡
*
* @return 结果状态集
*/
private CheckConsumerShardingResultContainer checkShardConsumerReBalanceStatus() {
// step1. 检查自身是否存在shard, 不存在则立即进行一次重分配（消费者机器数大于分片数时，重平衡动作将是无效动作）
// step2. 检查所有shard列表，是否有未被分配的shard,如有，立即触发一次重分配
// step3. 检查是否有负荷比较高的机器，如有触发平衡（功能预留，此功能需要基于统计信息）
CheckConsumerShardingResultContainer resultContainer = new CheckConsumerShardingResultContainer();
final List<String> activeServersList = shardAssignMaster.getAllOnlineServerList();
final List<String> allShardList = getAllShardList();
// 计算空闲机器
Map<String, Integer> hostConsumeLoadCountMap = new HashMap<>();
List<String> idleServerList = filterIdleServerList(activeServersList, hostConsumeLoadCountMap);
// 计算未被分配的shard
List<String> unAssignedShardList = filterUnAssignedShardList(allShardList);
// 根据资源信息，得出目前的负载状态
ReBalanceStatusResultEnum statusResult = computeReBalanceStatusOnResources(
unAssignedShardList, idleServerList, hostConsumeLoadCountMap);
resultContainer.setAllServerList(activeServersList);
resultContainer.setAllShardList(allShardList);
resultContainer.setIdleServerList(idleServerList);
resultContainer.setUnAssignedShardList(unAssignedShardList);
resultContainer.setServerConsumeShardLoad(hostConsumeLoadCountMap);
resultContainer.setStatusResultType(statusResult);
return resultContainer;
}
/**
* 根据给定资源信息，计算出目前的负载状态
*
* @param unAssignedShardList 未分配的shard列表
* @param idleServerList 空闲机器列表
* @param hostConsumeLoadMap 机器消费计数容器（负载情况）
* @return 状态值
*/
private ReBalanceStatusResultEnum computeReBalanceStatusOnResources(
List<String> unAssignedShardList,
List<String> idleServerList,
Map<String, Integer> hostConsumeLoadMap) {
// 没有未分配的shard，检测是否平衡即可
// 0. 有空闲机器，则直接分配给空闲机器即可
// 1. 最大消费shard-最小消费shard数 >= 2, 则说明有机器消费过多shard，需重分配
// 2. 机器负载平衡，无须调整
if(unAssignedShardList.isEmpty()) {
int minConsume = MAX_CONSUMER_SHARD_LOAD;
int maxConsume = 0;
for (Map.Entry<String, Integer> entry : hostConsumeLoadMap.entrySet()) {
int gotCount = entry.getValue();
if(gotCount > maxConsume) {
maxConsume = gotCount;
}
if(gotCount < minConsume) {
minConsume = gotCount;
}
}
// 因有未分配的机器，假如现有的机器消费都是2，则需要重分配的大压力的机器 shard 给空闲机器
if(!idleServerList.isEmpty()) {
if (maxConsume > 1) {
return ReBalanceStatusResultEnum.HEAVY_LOAD_WITH_CONSUMER_IDLE_BALANCE_NEEDED;
}
}
// 有消费相差2的机器，重新分配，从大数上借调到小数上
if(maxConsume > minConsume + 1) {
return ReBalanceStatusResultEnum.HEAVY_LOAD_BALANCE_NEEDED;
}
return ReBalanceStatusResultEnum.OK;
}
// 有可用shard
// 3. 有空闲机器,直接让空闲shard分配给这些空闲机器就ok了
// 4. 没有空闲机器，须将空闲shard 分配给负载小的机器
if(idleServerList.isEmpty()) {
return ReBalanceStatusResultEnum.UNASSIGNED_SHARD_WITHOUT_CONSUMER_IDLE_EXISTS;
}
return ReBalanceStatusResultEnum.UNASSIGNED_SHARD_WITH_CONSUMER_IDLE_EXISTS;
}
/**
* 过滤出空闲的机器列表
*
* @param activeServersList 所有机器列表
* @return 空闲机器集, 且将各自消费数放入计数容器
*/
private List<String> filterIdleServerList(List<String> activeServersList, Map<String, Integer> hostConsumeCountMap) {
List<String> idleServerList = new ArrayList<>();
for (String hostname1 : activeServersList) {
if(!shardAssignMaster.isConsumerServerAlive(hostname1)) {
shardAssignMaster.invalidateOfflineServer(hostname1);
continue;
}
int consumeCount;
Set<String> consumeShardSet = shardAssignMaster.getServerDutyConsumeShardSet(hostname1);
if(consumeShardSet == null || consumeShardSet.isEmpty()) {
idleServerList.add(hostname1);
consumeCount = 0;
}
else {
consumeCount = consumeShardSet.size();
}
hostConsumeCountMap.put(hostname1, consumeCount);
}
return idleServerList;
}
/**
* 过滤出未分配的shard列表
*
* @param allShardList 所有shard
* @return 未分配的shard
*/
private List<String> filterUnAssignedShardList(List<String> allShardList) {
List<String> unAssignedShardList = new ArrayList<>();
for (String shardId1 : allShardList) {
String consumeHostname = shardAssignMaster.getShardAssignedServer(shardId1);
// 如果不为空，则之前分配过，检查机器是否下线
// 如果为空，则是第一次分配
if(!StringUtils.isBlank(consumeHostname)) {
if(!shardAssignMaster.isConsumerServerAlive(consumeHostname)) {
// 清除下线机器信息,将当前shard置为空闲
shardAssignMaster.invalidateOfflineServer(consumeHostname);
shardAssignMaster.invalidateShardAssignInfo(shardId1);
unAssignedShardList.add(shardId1);
}
}
else {
unAssignedShardList.add(shardId1);
}
}
return unAssignedShardList;
}
/**
* 尝试获取协调者协调锁
*
* 在集群环境中，只允许有一个协调器在运行
*
* @return true:成功， false:失败，不得进行协调分配工作
*/
private boolean tryCoWorkerLock() {
return redisPoolUtil.getDistributedLock("distributedLock", HOST_NAME, 30);
}
/**
* 释放协调锁，以便下次再竞争
*/
private void releaseCoWorkerLock() {
redisPoolUtil.releaseDistributedLock("distributedLock", HOST_NAME);
}
/**
* 重新平衡消费者和shard的关系
*
* @param resultContainer 待重平衡状态
*/
private void reBalanceConsumerShard(CheckConsumerShardingResultContainer resultContainer) {
// 集群消费loghub数据动态伸缩策略，根据负载状态，调用相应策略进行重平衡
StatusReBalanceStrategy strategy = StatusReBalanceStrategyFactory.createStatusReBalanceAlgorithm(resultContainer, shardAssignMaster);
strategy.loadBalance();
}
/**
* 获取分片列表
*
* @return 分片列表，如: 0,1,2,3
*/
private List<String> getAllShardList() {
// 实时读取列表
List<String> shardList = Lists.newArrayList();
try {
ListShardResponse listShardResponse = mClient.ListShard(logHubProperties.getProjectName(),
logHubProperties.getEventlogStore());
ArrayList<Shard> getShards = listShardResponse.GetShards();
for (Shard shard : getShards) {
shardList.add(shard.GetShardId() + "");
}
}
catch (LogException e) {
logger.error("loghub 获取shard列表 error :", e);
}
return shardList;
}
}
如上，就是协调均衡主框架。主要逻辑如下：
1. 启动时初始化各种端，分配器，注册自己到控制中心等等;
2. 以线程的形式，被外部以定时任务执行的方式调用;
3. 检查任务前，须获得检查锁，否则直接返回;
4. 先获得目前机器的所有消费情况和shard的分配情况，得出资源负载数据;
5. 根据得到的数据信息,推算出目前的平衡状态;
6. 根据平衡状态，调用相应的平衡策略进行重平衡;
7. 等待下一次调度;
检查结果将作为后续选择均衡策略的依据，所以需要相应的状态容器保存。如下:
/**
* 集群状态预检查 结果容器
*/
class CheckConsumerShardingResultContainer {
/**
* 所有shard列表
*/
private List<String> allShardList;
/**
* 未被分配的shard列表
*/
private List<String> unAssignedShardList;
/**
* 所有机器列表
*/
private List<String> allServerList;
/**
* 空闲的机器列表（未被分配shard）
*/
private List<String> idleServerList;
/**
* 机器消费shard的负载计数容器
*/
private Map<String, Integer> serverConsumeShardLoad;
/**
* 状态检查结果类型
*/
private ReBalanceStatusResultEnum statusResultType;
public Map<String, Integer> getServerConsumeShardLoad() {
return serverConsumeShardLoad;
}
public void setServerConsumeShardLoad(Map<String, Integer> serverConsumeShardLoad) {
this.serverConsumeShardLoad = serverConsumeShardLoad;
}
public List<String> getAllShardList() {
return allShardList;
}
public void setAllShardList(List<String> allShardList) {
this.allShardList = allShardList;
}
public List<String> getUnAssignedShardList() {
return unAssignedShardList;
}
public void setUnAssignedShardList(List<String> unAssignedShardList) {
this.unAssignedShardList = unAssignedShardList;
}
public List<String> getAllServerList() {
return allServerList;
}
public void setAllServerList(List<String> allServerList) {
this.allServerList = allServerList;
}
public List<String> getIdleServerList() {
return idleServerList;
}
public void setIdleServerList(List<String> idleServerList) {
this.idleServerList = idleServerList;
}
public ReBalanceStatusResultEnum getStatusResultType() {
return statusResultType;
}
public void setStatusResultType(ReBalanceStatusResultEnum statusResultType) {
this.statusResultType = statusResultType;
}
}
针对多个平衡策略算法，使用一个工厂类来生产各种策略实例。如下:
/**
* 再平衡算法工厂类
*/
class StatusReBalanceStrategyFactory {
/**
* 无需做平衡的控制器
*/
private static final StatusReBalanceStrategy EMPTY_BALANCER = new EmptyReBalancer();
/**
* 根据当前的负载状态，创建对应的负载均衡算法
*
* @param resultContainer 负载状态集
* @param shardAssignMaster 分片分配管理者实例
* @return 算法实例
*/
public static StatusReBalanceStrategy createStatusReBalanceAlgorithm(CheckConsumerShardingResultContainer resultContainer, ShardAssignMaster shardAssignMaster) {
ReBalanceStatusResultEnum balanceStatus = resultContainer.getStatusResultType();
switch (balanceStatus) {
case OK:
return EMPTY_BALANCER;
case UNASSIGNED_SHARD_WITH_CONSUMER_IDLE_EXISTS:
return new UnAssignedShardWithConsumerIdleReBalancer(shardAssignMaster,
resultContainer.getUnAssignedShardList(), resultContainer.getIdleServerList());
case UNASSIGNED_SHARD_WITHOUT_CONSUMER_IDLE_EXISTS:
return new UnassignedShardWithoutConsumerIdleReBalancer(shardAssignMaster,
resultContainer.getUnAssignedShardList(), resultContainer.getServerConsumeShardLoad());
case HEAVY_LOAD_BALANCE_NEEDED:
return new HeavyLoadReBalancer(shardAssignMaster, resultContainer.getServerConsumeShardLoad());
case HEAVY_LOAD_WITH_CONSUMER_IDLE_BALANCE_NEEDED:
return new HeavyLoadWithConsumerIdleReBalancer(shardAssignMaster,
resultContainer.getServerConsumeShardLoad(), resultContainer.getIdleServerList());
default:
break;
}
return EMPTY_BALANCER;
}
}
/**
* 负载均衡策略统一接口
*/
interface StatusReBalanceStrategy {
/**
* 执行负载均衡方法
*/
public void loadBalance();
}
针对各种场景的负载均衡,各自实现如下。其中，无需操作时，将返回一个空操作实例！
1. 空操作实例
/**
* 无需做平衡的控制器
*
* @see ReBalanceStatusResultEnum#OK 状态枚举
*/
class EmptyReBalancer implements StatusReBalanceStrategy {
@Override
public void loadBalance() {
// ignore ...
}
}
2. 分配剩余shard给空闲的机器控制器
/**
* 为所有空闲的其他空闲机器分配可用 shard 的控制器
*
* @see ReBalanceStatusResultEnum#UNASSIGNED_SHARD_WITHOUT_CONSUMER_IDLE_EXISTS 状态枚举
*/
class UnAssignedShardWithConsumerIdleReBalancer implements StatusReBalanceStrategy {
/**
* 未被分配的分片列表
*/
private List<String> unAssignedShardList;
/**
* 分片分配管理者实例
*/
private ShardAssignMaster shardAssignMaster;
/**
* 空闲的机器列表
*/
private List<String> idleServerList;
public UnAssignedShardWithConsumerIdleReBalancer(
ShardAssignMaster shardAssignMaster,
List<String> unAssignedShardList,
List<String> idleServerList) {
this.shardAssignMaster = shardAssignMaster;
this.unAssignedShardList = unAssignedShardList;
this.idleServerList = idleServerList;
}
@Override
public void loadBalance() {
// 1. 找出还未被消费的shard
// 2. 依次分配给各空闲机器，每个空闲机器只至多分配一个shard
int serverIndex = 0;
for (String shard1 : unAssignedShardList) {
// 轮询分配shard, 先只给一个机器分配一个shard
if(serverIndex >= idleServerList.size()) {
break;
}
String serverHostname = idleServerList.get(serverIndex++);
shardAssignMaster.assignShardToServer(shard1, serverHostname);
}
}
}
3. 分配剩余shard给负载低的机器的控制器
/**
* 有空闲shard场景 的控制器 ， 须找出负载最低的机器塞入shard到现有的机器中（可能是有机器下线导致）
*
* @see ReBalanceStatusResultEnum#UNASSIGNED_SHARD_WITHOUT_CONSUMER_IDLE_EXISTS 状态枚举
*/
class UnassignedShardWithoutConsumerIdleReBalancer implements StatusReBalanceStrategy {
/**
* 未被分配分片列表
*/
private List<String> unAssignedShardList;
/**
* 分片管理者实例
*/
private ShardAssignMaster shardAssignMaster;
/**
* 消费者负载情况
*/
private Map<String, Integer> consumerLoadCount;
public UnassignedShardWithoutConsumerIdleReBalancer(
ShardAssignMaster shardAssignMaster,
List<String> unAssignedShardList,
Map<String, Integer> consumerLoadCount) {
this.shardAssignMaster = shardAssignMaster;
this.unAssignedShardList = unAssignedShardList;
this.consumerLoadCount = consumerLoadCount;
}
@Override
public void loadBalance() {
// 1. 找出负载最低的机器
// 2. 依次分配shard给该机器
// 3. 分配的后负载数+1, 循环分配
// 先根据空闲数，计算出一个可以接受新shard的机器的shard负载最低值，然后依次分配给这些机器
for (String shard1 : unAssignedShardList) {
// 按负载最小分配原则 分配shard
Map.Entry<String, Integer> minLoadServer = getMinLoadServer(consumerLoadCount);
String serverHostname = minLoadServer.getKey();
// 分配shard给机器
shardAssignMaster.assignShardToServer(shard1, serverHostname);
// 负载数 +1
minLoadServer.setValue(minLoadServer.getValue() + 1);
}
}
/**
* 获取负载最小的机器名备用
*
* @param loadCount 负载数据
* @return 最小负载机器
*/
private Map.Entry<String, Integer> getMinLoadServer(Map<String, Integer> loadCount) {
int minCount = MAX_CONSUMER_SHARD_LOAD;
Map.Entry<String, Integer> minLoadServer = null;
for(Map.Entry<String, Integer> server1 : loadCount.entrySet()) {
if(server1.getValue() < minCount) {
minCount = server1.getValue();
minLoadServer = server1;
}
}
return minLoadServer;
}
}
4. 将现有机器消费情况做重分配，从而使各自负载相近控制器
/**
* 负载不均衡导致的 重新均衡控制器，将消费shard多的机器的 shard 拆解部分到 消费少的机器上 （须上锁）
*
* @see ReBalanceStatusResultEnum#HEAVY_LOAD_BALANCE_NEEDED 状态枚举
*/
class HeavyLoadReBalancer implements StatusReBalanceStrategy {
/**
* 分片分配管理者实例
*/
private ShardAssignMaster shardAssignMaster;
/**
* 机器消费负载情况
*/
private Map<String, Integer> consumerLoadCount;
public HeavyLoadReBalancer(ShardAssignMaster shardAssignMaster, Map<String, Integer> consumerLoadCount) {
this.shardAssignMaster = shardAssignMaster;
this.consumerLoadCount = consumerLoadCount;
}
@Override
public void loadBalance() {
// 1. 找出所有机器的消费数的平均线值
// 2. 负载数大于均线1的，直接抽出多余的shard, 放到待分配容器中
// 3. 从大到小排序负载机器
// 4. 从大的负载上减少shard到最后的机器上，直到小的机器达到平均负载线最贴近的地方，或者小的机器到达平均负载线最贴近的地方
// 5. ++大负载机器 或者 --小负载机器，下一次循环
double avgLoadCount = computeAliveServersAvgLoadCount(consumerLoadCount);
List<Map.Entry<String, Integer>> sortedLoadCountList = sortLoadCountByLoadWithSmallEndian(consumerLoadCount);
int bigLoadIndex = 0;
int smallLoadIndex = sortedLoadCountList.size() - 1;
for (;;) {
// 首先检测是否已遍历完成，完成后不再进行分配
if(isRoundRobinComplete(bigLoadIndex, smallLoadIndex)) {
break;
}
Map.Entry<String, Integer> bigLoadServerEntry = sortedLoadCountList.get(bigLoadIndex);
double canTakeCountFromBigLoad = bigLoadServerEntry.getValue() - avgLoadCount;
if(canTakeCountFromBigLoad < 1) {
bigLoadIndex += 1;
continue;
}
for (int reAssignShardIndex = 0;
reAssignShardIndex < canTakeCountFromBigLoad; reAssignShardIndex++) {
if(isRoundRobinComplete(bigLoadIndex, smallLoadIndex)) {
break;
}
Map.Entry<String, Integer> smallLoadServerEntry = sortedLoadCountList.get(smallLoadIndex);
double canPutIntoSmallLoad = avgLoadCount - smallLoadServerEntry.getValue();
if(canPutIntoSmallLoad < 1) {
smallLoadIndex -= 1;
continue;
}
// 此处可以使用管道操作，更流畅, 或者更准确的说，使用事务操作
// 从 bigLoad 中移除shard 0
// 将移除的 shard 上锁，以防后续新机器立即消费，导致数据异常
// 添加新shard到 smallLoad 中
String firstLoadSHardId = shardAssignMaster.popServerFirstConsumeShardId(bigLoadServerEntry.getKey());
bigLoadServerEntry.setValue(bigLoadServerEntry.getValue() - 1);
// 上锁分片，禁用消费
shardAssignMaster.lockShardId(firstLoadSHardId);
// 添加shard到 smallLoad 中
shardAssignMaster.assignShardToServer(firstLoadSHardId, smallLoadServerEntry.getKey());
smallLoadServerEntry.setValue(smallLoadServerEntry.getValue() + 1);
}
}
}
/**
* 判定轮询是否完成
*
* @param startIndex 开始下标
* @param endIndex 结束下标
* @return true: 轮询完成, false: 未完成
*/
private boolean isRoundRobinComplete(int startIndex, int endIndex) {
return startIndex == endIndex;
}
/**
* 从大到小排序 负载机器
*
* @param consumerLoadCount 总负载情况
* @return 排序后的机器列表
*/
private List<Map.Entry<String, Integer>> sortLoadCountByLoadWithSmallEndian(Map<String, Integer> consumerLoadCount) {
List<Map.Entry<String, Integer>> sortedList = new ArrayList<>(consumerLoadCount.entrySet());
sortedList.sort(new Comparator<Map.Entry<String, Integer>>() {
@Override
public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
return o2.getValue() - o1.getValue();
}
});
return sortedList;
}
/**
* 计算平均每台机器的消费shard负载
*
* @param loadCount 总负载指标容器
* @return 负载均线
*/
private double computeAliveServersAvgLoadCount(Map<String, Integer> loadCount) {
int totalServerCount = loadCount.size();
int totalShardCount = 0;
for(Integer consumeShardCount : loadCount.values()) {
totalShardCount += consumeShardCount;
}
return (double) totalShardCount / totalServerCount;
}
}
5. 从负载重的机器上剥夺shard,分配给空闲的机器 控制器
/**
* 负载不均衡，且存在空闲的机器, 此时应是 均值与最大值之间相差较小值，但是至少有一个 消费2 的机器，可以剥夺其1个shard给空闲机器 的控制器
*
* @see ReBalanceStatusResultEnum#HEAVY_LOAD_WITH_CONSUMER_IDLE_BALANCE_NEEDED 状态枚举
*/
class HeavyLoadWithConsumerIdleReBalancer implements StatusReBalanceStrategy {
/**
* 分片分配管理者实例
*/
private ShardAssignMaster shardAssignMaster;
/**
* 空闲的机器列表
*/
private List<String> idleServerList;
/**
* 机器消费负载情况
*/
private Map<String, Integer> consumerLoadCount;
public HeavyLoadWithConsumerIdleReBalancer(
ShardAssignMaster shardAssignMaster,
Map<String, Integer> consumerLoadCount,
List<String> idleServerList) {
this.shardAssignMaster = shardAssignMaster;
this.consumerLoadCount = consumerLoadCount;
this.idleServerList = idleServerList;
}
@Override
public void loadBalance() {
// 1. 找出还未被消费的shard
// 2. 分配一个给自己
// 3. 如果还有其他机器也未分配，则同样进行分配
for (String idleHostname1 : idleServerList) {
Map.Entry<String, Integer> maxLoadEntry = getMaxLoadConsumerEntry(consumerLoadCount);
// 本身只有一个则不再分配负担了
if(maxLoadEntry.getValue() <= 1) {
break;
}
String maxLoadServerHostname = maxLoadEntry.getKey();
// 此处可以使用管道操作，更流畅, 或者更准确的说，使用事务操作
// 从 bigLoad 中移除shard 0
// 将移除的 shard 上锁，以防后续新机器立即消费，导致数据异常
// 添加新shard到 smallLoad 中
String firstLoadSHardId = shardAssignMaster.popServerFirstConsumeShardId(maxLoadServerHostname);
maxLoadEntry.setValue(maxLoadEntry.getValue() - 1);
// 上锁卸载下来的shard,锁定50s
shardAssignMaster.lockShardId(firstLoadSHardId);
// 添加shard到 smallLoad 中
shardAssignMaster.assignShardToServer(firstLoadSHardId, idleHostname1);
consumerLoadCount.put(idleHostname1, 1);
}
}
/**
* 获取负载最大的机器实例作
*
* @param consumerLoadCount 所有机器的负载情况
* @return 最大负载机器实例
*/
private Map.Entry<String, Integer> getMaxLoadConsumerEntry(Map<String, Integer> consumerLoadCount) {
Integer maxConsumeCount = 0;
Map.Entry<String, Integer> maxEntry = null;
for (Map.Entry<String, Integer> server1 : consumerLoadCount.entrySet()) {
if(server1.getValue() > maxConsumeCount) {
maxConsumeCount = server1.getValue();
maxEntry = server1;
}
}
return maxEntry;
}
}
如上，各个平衡策略，实现各自的功能，就能掌控整个集群的消费控制了！
除了上面的主料，还有一些附带的东西！
【2】均衡状态枚举值如下:
/**
* 再平衡检测结果类型枚举
*
*/
public enum ReBalanceStatusResultEnum {
/**
* 一切正常，无须操作
*/
OK("一切正常，无须操作"),
/**
* 有新下线机器，可以将其分片分配给其他机器
*/
UNASSIGNED_SHARD_WITHOUT_CONSUMER_IDLE_EXISTS("有未分配的分片，可以分配给其他机器"),
/**
* 有未分配的分片，且有空闲机器，直接将空闲shard分配给空闲机器即可（最好只分配1个，以便其他机器启动后可用）
*/
UNASSIGNED_SHARD_WITH_CONSUMER_IDLE_EXISTS("有未分配的分片，且有空闲机器"),
/**
* 负载不均衡，须生平衡
*/
HEAVY_LOAD_BALANCE_NEEDED("负载不均衡，须生平衡"),
/**
* 负载不均衡，且存在空闲的机器, 此时应是 均值与最大值之间相差较小值，但是至少有一个 消费2 的机器，可以剥夺其1个shard给空闲机器
*/
HEAVY_LOAD_WITH_CONSUMER_IDLE_BALANCE_NEEDED("负载不均衡，且存在空闲的机器"),
;
private ReBalanceStatusResultEnum(String remark) {
// ignore
}
}
【3】RedisKeyConstants 常量定义
/**
* redis 相关常量定义
*/
public class RedisKeyConstants {
/**
* 在线机器缓存key.与心跳同时作用
*
* @see #SERVER_ALIVE_HEARTBEAT_CACHE_PREFIX
*/
public static final String ALL_ONLINE_SERVER_CACHE_KEY = "prefix:active.servers";
/**
* 机器消费shard情况 缓存key前缀
*/
public static final String SERVER_CONSUME_CACHE_KEY_PREFIX = "prefix:log.consumer:server:";
/**
* 分片被分配情况 缓存key前缀
*/
public static final String SHARD_ASSIGNED_CACHE_KEY_PREFIX = "prefix:shard.assigned:id:";
/**
* 分片锁 缓存key前缀， 当上锁时，任何机器不得再消费
*/
public static final String SHARD_LOCK_CONSUME_CACHE_PREFIX = "prefix:consume.lock.shard:id:";
/**
* 存活机器心跳，与上面的机器形成呼应
*
* @see #ALL_ONLINE_SERVER_CACHE_KEY
*/
public static final String SERVER_ALIVE_HEARTBEAT_CACHE_PREFIX = "prefix:log.consumer:server.heartbeat:";
/**
* 单个消费者最大消费负载数 （一个不可能达到的值）
*/
public static final Integer MAX_CONSUMER_SHARD_LOAD = 9999;
}
【4】shard分配控制器负责所有shard分配
import com.test.utils.RedisPoolUtil;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
/**
* shard分配管理者 （尽量使用接口表达）
*
*/
public class ShardAssignMaster {
private RedisPoolUtil redisPoolUtil;
public ShardAssignMaster(RedisPoolUtil redisPoolUtil) {
this.redisPoolUtil = redisPoolUtil;
}
/**
* 注册消费者到 控制中心（注册中心）
*/
public void registerConsumer(String serverHostname) {
// 注册server到 redis zset 中，如有条件，可以使用 zk 进行操作，也许更好
redisPoolUtil.zadd(RedisKeyConstants.ALL_ONLINE_SERVER_CACHE_KEY, (double)System.currentTimeMillis(), serverHostname);
}
/**
* 心跳发送数据
*/
public void sendHeartbeat(String serverHostname) {
String heartbeatCacheKey = RedisKeyConstants.SERVER_ALIVE_HEARTBEAT_CACHE_PREFIX + serverHostname;
redisPoolUtil.set(heartbeatCacheKey, "1", 30);
}
/**
* 检测指定消费者服务器还存活与否
*
* @param consumeHostname 机器名
* @return true: 存活, false: 宕机
*/
public boolean isConsumerServerAlive(String consumeHostname) {
String aliveValue = redisPoolUtil.get(RedisKeyConstants.SERVER_ALIVE_HEARTBEAT_CACHE_PREFIX + consumeHostname);
return aliveValue != null
&& "1".equals(aliveValue);
}
/**
* 获取并删除指定server的所属消费的第一个 shardId
*
* @param serverHostname 机器名
* @return 第一个shardId
*/
public String popServerFirstConsumeShardId(String serverHostname) {
String bigLoadConsumerServerCacheKey = RedisKeyConstants.SERVER_CONSUME_CACHE_KEY_PREFIX + serverHostname;
Set<String> firstLoadShardSet = redisPoolUtil.zrange(bigLoadConsumerServerCacheKey, 0, 0);
String firstLoadSHardId = firstLoadShardSet.iterator().next();
redisPoolUtil.zrem(bigLoadConsumerServerCacheKey, firstLoadSHardId);
redisPoolUtil.expire(bigLoadConsumerServerCacheKey, 60);
return firstLoadSHardId;
}
/**
* 对shard进行上锁，禁止所有消费行为
*
* @param shardId 分片id
*/
public void lockShardId(String shardId) {
String shardLockCacheKey = RedisKeyConstants.SHARD_LOCK_CONSUME_CACHE_PREFIX + shardId;
redisPoolUtil.set(shardLockCacheKey, "1", 50);
}
/**
* 分配shard分片数据给 指定server
*
* @param shardId 分片id
* @param serverHostname 分配给的消费者机器名
*/
public void assignShardToServer(String shardId, String serverHostname) {
String smallLoadConsumerServerCacheKey = RedisKeyConstants.SERVER_CONSUME_CACHE_KEY_PREFIX + serverHostname;
redisPoolUtil.zadd(smallLoadConsumerServerCacheKey, (double)System.currentTimeMillis(), shardId);
redisPoolUtil.expire(smallLoadConsumerServerCacheKey, 60);
// 更新新的shard消费者标识
String shardIdAssignCacheKey = RedisKeyConstants.SHARD_ASSIGNED_CACHE_KEY_PREFIX + shardId;
redisPoolUtil.set(shardIdAssignCacheKey, serverHostname);
}
/**
* 获取被分配了shardId的server信息
*
* @param shardId 要检查的分片id
* @return 被分配了shardId 的机器名
*/
public String getShardAssignedServer(String shardId) {
String shardAssignCacheKey = RedisKeyConstants.SHARD_ASSIGNED_CACHE_KEY_PREFIX + shardId;
return redisPoolUtil.get(shardAssignCacheKey);
}
/**
* 删除shard的分配信息，使无效化
*
* @param shardId 要删除的分片id
*/
public void invalidateShardAssignInfo(String shardId) {
String shardAssignCacheKey = RedisKeyConstants.SHARD_ASSIGNED_CACHE_KEY_PREFIX + shardId;
redisPoolUtil.del(shardAssignCacheKey);
}
/**
* 清理下线机器
*
* @param hostname 下线机器名
*/
public void invalidateOfflineServer(String hostname) {
redisPoolUtil.zrem(RedisKeyConstants.ALL_ONLINE_SERVER_CACHE_KEY, hostname);
}
/**
* 获取机器消费的shard列表
*
* @param serverHostname 机器主机名
* @return shard列表 或者 null
*/
public Set<String> getServerDutyConsumeShardSet(String serverHostname) {
String serverDutyConsumeShardCacheKey = RedisKeyConstants.SERVER_CONSUME_CACHE_KEY_PREFIX + serverHostname;
return redisPoolUtil.zrange(serverDutyConsumeShardCacheKey, 0, -1);
}
/**
* 获取所有在线机器列表
*
* @return 在线机器列表
*/
public List<String> getAllOnlineServerList() {
Set<String> hostnameSet = redisPoolUtil.zrange(RedisKeyConstants.ALL_ONLINE_SERVER_CACHE_KEY, 0, -1);
return new ArrayList<>(hostnameSet);
}
}
以上是协同负载均衡器代码实现。
【5】当然你还需要一个消费者
接下来我们还要看下消费者如何实现消费。
import com.test.utils.RedisPoolUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
* 消费者业务线程
*
*/
public class LoghubConsumeWorker implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(LoghubConsumeWorker.class);
private RedisPoolUtil redisPoolUtil;
private String HOST_NAME;
/**
* 因消费者数目不一定，所以使用 CachedThreadPool
*/
private ExecutorService consumeExecutorService = Executors.newCachedThreadPool();
public LoghubConsumeWorker(RedisPoolUtil redisPoolUtil) {
this(redisPoolUtil, null);
}
public LoghubConsumeWorker(RedisPoolUtil redisPoolUtil, String hostName) {
this.redisPoolUtil = redisPoolUtil;
// 为测试需要添加的 hostName
HOST_NAME = hostName;
initSharedVars();
}
/**
* 初始化公共变量
*/
private void initSharedVars() {
try {
if(HOST_NAME != null) {
return;
}
HOST_NAME = InetAddress.getLocalHost().getHostName();
}
catch (UnknownHostException e) {
throw new RuntimeException("init error : 获取服务器主机名失败");
}
}
@Override
public void run() {
while (!Thread.interrupted()) {
// 先获取所有分配给的shard列表，为空则进入下一次循环（注意此时阻塞锁不能起作用）
Set<String> shardsSet = blockingTakeAvailableConsumeShardList();
try {
// 消费所有给定shard数据
consumeLogHubShards(shardsSet);
} catch (Exception e) {
logger.error("消费loghub, error", e);
}
}
}
/**
* 获取可用的分片列表（没有则阻塞等待）
*
* @return 分片列表
*/
private Set<String> blockingTakeAvailableConsumeShardList() {
while (!Thread.interrupted()) {
String serverConsumeCacheKey = RedisKeyConstants.SERVER_CONSUME_CACHE_KEY_PREFIX + HOST_NAME;
Set<String> shardsSet = redisPoolUtil.zrange(serverConsumeCacheKey, 0, -1);
if (shardsSet != null && !shardsSet.isEmpty()) {
return shardsSet;
}
logger.warn(" =========== 当前主机[hostname:{}]未查询到任何shard =========", HOST_NAME);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
logger.error("LogHubClientWork run 未获取到该主机的shard时,每隔1秒钟获取 ,error : {}", e);
}
}
return null;
}
/**
* 消费loghub 分片数据
*
* @param shardsSet 被分配的分片列表
*/
public void consumeLogHubShards(Set<String> shardsSet) throws InterruptedException {
if(shardsSet == null || shardsSet.isEmpty()) {
return;
}
// 此处使用 CountdownLatch, 保证至少有一个任务完成时，才开始下一次任务的调入
// Semaphore semaphoreLock = new Semaphore(shardsSet.size());
CountDownLatch openDoorLatch = new CountDownLatch(1);
boolean startNewJobAtLeastOnce = false;
for (String shard : shardsSet) {
// 检测当前shard是否处于锁定状态，如果锁定则不能消费, 注意锁情况
if(isShardLocked(shard)) {
logger.info("=============== shard:{} is locked, continue... ======", shard);
continue;
}
int shardId = Integer.parseInt(shard);
LoghubConsumerTaskExecutor consumer = getConsumerExecutor(shardId);
// consumer 应保证有所消费，如果没有消费，则自行等待一个长周期，外部应只管调入请求
// consumer 应保证所有消费，在上一个任务未完成时，不得再开启下一轮提交消费
boolean startNewJob = consumer.startNewConsumeJob(openDoorLatch);
if(startNewJob) {
// start failed, prev job is running maybe
// ignore job, no blocking
startNewJobAtLeastOnce = true;
}
}
// 任意一个任务完成，都将打开新的分配周期,且后续 countDown 将无效,此处可能导致死锁
if(startNewJobAtLeastOnce) {
openDoorLatch.await();
}
else {
// 当本次分配调度一个任务都未提交时，则睡眠等待
// （一般此情况为 消费者被分配了上了锁的shard时，即抢占另的机器的shard, 需要给别的机器备份数据时间锁）
Thread.sleep(200);
}
}
/**
* 检测分片是否被锁定消费了
*
* @param shardId 分片id
* @return true:锁定, false:未锁定可用
*/
private boolean isShardLocked(String shardId) {
String shardCacheKey = RedisKeyConstants.SHARD_LOCK_CONSUME_CACHE_PREFIX + shardId;
String lockValue = redisPoolUtil.get(shardCacheKey);
return !StringUtils.isBlank(lockValue)
&& "1".equals(lockValue);
}
/**
* 获取消费者实例，针对一个shard, 只创建一个实例
*/
private Map<Integer, LoghubConsumerTaskExecutor> mShardConsumerMap = new ConcurrentHashMap<>();
private LoghubConsumerTaskExecutor getConsumerExecutor(final int shardId) {
LoghubConsumerTaskExecutor consumer = mShardConsumerMap.get(shardId);
if (consumer != null) {
return consumer;
}
consumer = new LoghubConsumerTaskExecutor(new SingleShardConsumerJob(shardId));
mShardConsumerMap.put(shardId, consumer);
logger.info(" ======================= create new consumer executor shard:{}", shardId);
return consumer;
}
/**
* 消费者调度器
*
* 统一控制消费者的运行状态管控
*/
class LoghubConsumerTaskExecutor {
private Future<?> future;
private ConsumerJob consumerJob;
public LoghubConsumerTaskExecutor(ConsumerJob consumerJob) {
this.consumerJob = consumerJob;
}
/**
* 启动一个新消费任务
*
* @return true: 启动成功, false: 启动失败有未完成任务在前
*/
public boolean startNewConsumeJob(CountDownLatch latch) {
if(future == null
|| future.isCancelled() || future.isDone()) {
//没有任务或者任务已取消或已完成 提交任务
future = consumeExecutorService.submit(new Runnable() {
@Override
public void run() {
try {
consumerJob.consumeShardData();
}
finally {
latch.countDown();
}
}
});
return true;
}
return false;
}
}
}
/**
* 消费者任务接口定义
*/
interface ConsumerJob {
/**
* 消费数据具体逻辑实现
*/
public void consumeShardData();
}
/**
* 单个shard消费的任务实现
*/
class SingleShardConsumerJob implements ConsumerJob {
/**
* 当前任务的消费 shardId
*/
private int shardId;
public SingleShardConsumerJob(int shardId) {
this.shardId = shardId;
}
@Override
public void consumeShardData() {
System.out.println(LocalDateTime.now() + " - host -> consume shard: " + shardId);
try {
// do complex biz
// 此处如果发现shard 不存在异常，则应回调协调器，进行shard的移除
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
}
}
}
【6】当然你还需要一个demo
看不到效果，我就是不信！
所以来看个 demo 吧！
我们使用单机开多个 单元测试用例，直接测试就好！
测试代码:.
import com.test.common.config.LogHubProperties;
import com.test.utils.RedisPoolUtil;
import org.junit.Test;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 临时测试 负载均衡
*
*/
public class ShardConsumerLoadBalanceTest {
public static void main(String[] args) throws IOException {
startAConsumer();
System.in.read();
}
// 启动一个单元测试，就相当于启动一个消费者应用
@Test
public void mainMock() throws IOException {
startAConsumer();
System.in.read();
}
// 启动一个单元测试，就相当于启动一个消费者应用
@Test
public void startNewConsumer() throws IOException {
startAConsumer();
System.in.read();
}
// 启动一个单元测试，就相当于启动一个消费者应用
@Test
public void startNewConsumer2() throws IOException {
startAConsumer();
System.in.read();
}
private static void startAConsumer() {
RedisPoolUtil redisPoolUtil = new RedisPoolUtil();
redisPoolUtil.setIp("127.0.0.1");
redisPoolUtil.setMaxActive(111);
redisPoolUtil.setMaxIdle(1000);
redisPoolUtil.setPort(6379);
redisPoolUtil.setMaxWait(100000);
redisPoolUtil.setTimeout(100000);
redisPoolUtil.setPassWord("123");
redisPoolUtil.setDatabase(0);
redisPoolUtil.initPool();
LogHubProperties logHubProperties = new LogHubProperties();
logHubProperties.setProjectName("test");
logHubProperties.setEndpoint("cn-shanghai-finance-1.log.aliyuncs.com");
logHubProperties.setAccessKey("xxxx");
logHubProperties.setAccessKeyId("11111");
// 使用随机 hostname 模拟多台机器调用
Random random = new Random();
String myHostname = "my-host-" + random.nextInt(10);
// 启动管理线程
LoghubConsumerShardCoWorker shardCoWorker = new LoghubConsumerShardCoWorker(redisPoolUtil, logHubProperties, myHostname);
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
scheduledExecutorService.scheduleAtFixedRate(shardCoWorker, 5, 30, TimeUnit.SECONDS);
// 启动业务线程
ExecutorService executorService = Executors.newFixedThreadPool(2);
LoghubConsumeWorker worker = new LoghubConsumeWorker(redisPoolUtil, myHostname);
executorService.submit(worker);
}
}
如上，就可以实现自己的负载均衡消费了。
比如: 总分片数为4。
1. 最开始启动1个机器时，将会被分配 0,1,2,3。
2. 启动两个后，将分为 0,1; 2,3;
3. 启动3个后，将分为 0; 1; 2,3;
4. 反之，关闭一个机器后，将把压力分担到原机器上。
当做负载重分配时，将有50秒的锁定时间备份。
【7】待完善的点
当shard缩减时，应能够自动发现，从而去除原有的机器消费分配。而不是让消费者报错。
老话： 可以适当造轮子！
标签：机器,String,自定义,shard,private,kafka,loghub,return,public 来源： https://www.cnblogs.com/yougewe/p/11114348.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。