自己动手实现一个loghub(或kafka)分片消费负载均衡器

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;    }
}

/** * 集群状态预检查 结果容器 */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();}

/** * 无需做平衡的控制器 * * @see ReBalanceStatusResultEnum#OK 状态枚举 */class EmptyReBalancer implements StatusReBalanceStrategy {    @Override    public void loadBalance() {        // ignore ...    }}

/** * 为所有空闲的其他空闲机器分配可用 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);        }    }}

/** * 有空闲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;    }}

/** * 负载不均衡导致的 重新均衡控制器，将消费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;    }}

/** *  负载不均衡，且存在空闲的机器, 此时应是 均值与最大值之间相差较小值，但是至少有一个 消费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;    }}

/** * 再平衡检测结果类型枚举 * */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    }}

/** * 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;}

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);    }
}

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();        }    }}

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);
    }}