走过路过不要错过
点击蓝字关注我们
一般地,像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);
}
@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();
}
针对各种场景的负载均衡,各自实现如下。其中,无需操作时,将返回一个空操作实例!
/**
* 无需做平衡的控制器
*
* @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;
}
}
如上,各个平衡策略,实现各自的功能,就能掌控整个集群的消费控制了!
除了上面的主料,还有一些附带的东西!
【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】待完善的点
本文是基于loghub实现的分片拉取,其实在这方面loghub与kafka是如出一辙的,只是loghub更商业产品化。
当shard缩减时,应能够自动发现,从而去除原有的机器消费分配。而不是让消费者报错。
注意进行再均衡时,消费者偏移量问题,尤其是你为了性能使用了jvm本地变量保存偏移时,注意刷新该变量偏移。本文没有实现类似zookeeper强大的watch监听功能,但是有一个上锁等待的过程,你可以基于这个锁做一些力所能及的事!
老话:可以适当造轮子!