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