自己动手实现一个loghub(或kafka)分片消费负载均衡器
一般地，像kafka之类的消息中间件，作为一个可以保持历史消息的组件，其消费模型一般是主动拉取方式。这是为了给消费者足够的自由，回滚或者前进。
然而，也正是由于将消费消息的权力交给了消费者，所以，消费者往往需要承担更多的责任。比如：需要自行保存消费偏移量，以便后续可以知道从哪里继续。而当这一点处理不好时，则可能带来一些麻烦。
不管怎么样，解决方案也都是现成的，咱们也不用担心。
今天我们要谈论的是一个场景：如何让n个机器消费m个分片数据？（带状态的，即不能任意机器消费任意shard）
这在消息中间件的解决方案里，明白地写着，使用消费者群组就可以实现了。具体来说就是，每个分片至多会被一机器消费，每个机器则可以消费多个分片数据。即机器数据小于分片数时，分片会被均衡地分配到消费者中。当机器数大于分片数时，多余的机器将不做任何事情。
好吧，既然官方已经说明白了，那咱们应该就不再需要自己搞一个轮子了吧。
但是，我还有个场景：如果我要求在机器做负载重平衡时，需要保证被抽取出去的机器分片，至少保留一段时间，不允许任何机器消费该分片，因为可能还有数据需要备份。
针对这种场景，我想官方也许是有提供回调函数之类的解决方案的吧。不管了，反正我没找到，只能自己先造个轮子了。
本文场景前提:
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);}@Overridepublic 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);// 计算未被分配的shardList<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 {@Overridepublic 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;}@Overridepublic void loadBalance() {// 1. 找出还未被消费的shard// 2. 依次分配给各空闲机器，每个空闲机器只至多分配一个shardint serverIndex = 0;for (String shard1 : unAssignedShardList) {// 轮询分配shard, 先只给一个机器分配一个shardif(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;}@Overridepublic void loadBalance() {// 1. 找出负载最低的机器// 2. 依次分配shard给该机器// 3. 分配的后负载数+1, 循环分配// 先根据空闲数，计算出一个可以接受新shard的机器的shard负载最低值，然后依次分配给这些机器for (String shard1 : unAssignedShardList) {// 按负载最小分配原则 分配shardMap.Entry<String, Integer> minLoadServer = getMinLoadServer(consumerLoadCount);String serverHostname = minLoadServer.getKey();// 分配shard给机器shardAssignMaster.assignShardToServer(shard1, serverHostname);// 负载数 +1minLoadServer.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;}@Overridepublic 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>>() {@Overridepublic 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;}@Overridepublic 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,锁定50sshardAssignMaster.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;// 为测试需要添加的 hostNameHOST_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 : 获取服务器主机名失败");}}@Overridepublic 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 blockingstartNewJobAtLeastOnce = 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() {@Overridepublic 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;}@Overridepublic 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();}// 启动一个单元测试，就相当于启动一个消费者应用@Testpublic void mainMock() throws IOException {startAConsumer();System.in.read();}// 启动一个单元测试，就相当于启动一个消费者应用@Testpublic void startNewConsumer() throws IOException {startAConsumer();System.in.read();}// 启动一个单元测试，就相当于启动一个消费者应用@Testpublic 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】待完善的点
本文是基于loghub实现的分片拉取，其实在这方面loghub与kafka是如出一辙的，只是loghub更商业产品化。
当shard缩减时，应能够自动发现，从而去除原有的机器消费分配。而不是让消费者报错。
注意进行再均衡时，消费者偏移量问题，尤其是你为了性能使用了jvm本地变量保存偏移时，注意刷新该变量偏移。本文没有实现类似zookeeper强大的watch监听功能，但是有一个上锁等待的过程，你可以基于这个锁做一些力所能及的事！
老话：可以适当造轮子！
腾讯、阿里、滴滴后台面试题汇总总结 — （含答案）
面试：史上最全多线程面试题 ！
最新阿里内推Java后端面试题
JVM难学？那是因为你没认真看完这篇文章
关注作者微信公众号 —《JAVA烂猪皮》
了解更多java后端架构知识以及最新面试宝典
看完本文记得给作者点赞+在看哦~~~大家的支持，是作者源源不断出文的动力
作者：等你归去来
出处：https://www.cnblogs.com/yougewe/p/11114348.html
评论