重复消费自不必说，你 ClientID 都相同了。本篇着重聊聊为什么会消息堆积。
文章中讲到，初始化 Consumer 时，会初始化 Rebalance 的策略。你可以大致将 Rebalance 策略理解为如何将一个 Topic 下的 m 个 MessageQueue 分配给一个 ConsumerGroup 下的 n 个 Consumer 实例的策略，看着有些绕，其实就长这样：
而从 Consumer 初始化的源码中可以看出，默认情况下 Consumer 采取的 Rebalance 策略是 
AllocateMessageQueueAverage()。
默认的策略很好理解，将 MessageQueue 平均的分配给 Consumer。举个例子，假设有 8 个 MessageQueue，2 个 Consumer，那么每个 Consumer 就会被分配到 4 个 MessageQueue。
那如果分配不均匀怎么办？例如只有 7 个 MessageQueue，但是 Consumer 仍然是 2 个。此时 RocketMQ 会将多出来的部分，对已经排好序的 Consumer 再做平均分配，一个一个分发给 Consumer，直到分发完。例如刚刚说的 7 个 MessageQueue 和 2 个 ConsumerGroup 这种 case，排在第一个的 Consumer 就会被分配到 4 个 MessageQueue，而第二个会被分配到 3 个 MessageQueue。
大家可以先理解一下 AllocateMessageQueueAveragely 的实现，作为默认的 Rebalance 的策略，其实现位于这里：
接下来我们看看，AllocateMessageQueueAveragely 内部具体都做了哪些事情。
其核心其实就是实现的 AllocateMessageQueueStrategy 接口中的 allocate 方法。实际上，RocketMQ 对该接口总共有 5 种实现：
AllocateMessageQueueAveragely 只是其中的一种实现而已，那执行 allocate 它需要什么参数呢？
需要以下四个：
AllocateMessageQueueStrategy 的不同实现。
接下来我们就来看看 AllocateMessageQueueAveragely 是如何对 MessageQueue 进行分配的，之前讲源码我一般都会一步一步的来，结合源码跟图，但是这个源码太短了，我就直接先给出来吧。
public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
if (currentCID == null || currentCID.length() < 1) {
throw new IllegalArgumentException("currentCID is empty");
}
if (mqAll == null || mqAll.isEmpty()) {
throw new IllegalArgumentException("mqAll is null or mqAll empty");
}
if (cidAll == null || cidAll.isEmpty()) {
throw new IllegalArgumentException("cidAll is null or cidAll empty");
}
List<MessageQueue> result = new ArrayList<MessageQueue>();
// 判断一下当前的客户端是否在 cidAll 的集合当中
if (!cidAll.contains(currentCID)) {
log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
consumerGroup,
currentCID,
cidAll);
return result;
}
// 拿到当前消费者在所有的消费者实例数组中的位置
int index = cidAll.indexOf(currentCID);
// 用 messageQueue 的数量 对 消费者实例的数量取余数, 这个实际上就把不够均匀分的 MessageQueue 的数量算出来了
// 举个例子, 12 个 MessageQueue, 有 5 个 Consumer, 12 % 5 = 2
int mod = mqAll.size() % cidAll.size();
int averageSize =
mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size());
int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
int range = Math.min(averageSize, mqAll.size() - startIndex);
for (int i = 0; i < range; i++) {
result.add(mqAll.get((startIndex + i) % mqAll.size()));
}
return result;
}
其实前半部分都是些常规的 check，可以忽略不看，从这里：
int index = cidAll.indexOf(currentCID);
开始，才是核心逻辑。为了避免逻辑混乱，还是假设有 12 个 MessageQueue，5 个 Consumer，同时假设 index=0 。
那么 mod 的值就为 12 % 5 = 2 了。
而 averageSize 的值，稍微有点绕。如果 MessageQueue 的数量比消费者的数量还少，那么就为 1 ；否则，就走这一堆逻辑(mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size())。我们 index 是 0，而 mod 是 2，index < mod 则是成立的，那么最终 averageSize 的值就为 12 / 5 + 1 = 3。
接下来是 startIndex，由于这个三元运算符的条件是成立的，所以其值为 0 * 3 ，就为 0。
看了一大堆逻辑，是不是已经晕了？直接举实例：
12 个 Message Queue
5 个 Consumer 实例
按照上面的分法：
排在第 1 的消费者 分到 3 个
排在第 2 的消费者 分到 3 个
排在第 3 的消费者 分到 2 个
排在第 4 的消费者 分到 2 个
排在第 5 的消费者 分到 2 个
所以，你可以大致认为：
先“均分”，12 / 5 取整为 2。然后“均分”完之后还剩下 2 个，那么就从上往下，挨个再分配，这样第 1、第 2 个消费者就会被多分到 1 个。 所以如果有 13 个 MessageQueue，5 个 Consumer，那么第 1、第 2、第 3 就会被分配 3 个。但并不准确，因为分配的 MessageQueue 是一次性的，例如那 3 个 MessageQueue 是一次性获取的，不会先给 2 个，再给 1 个。 而我们开篇提到的 Consumer 的 ClientID 相同，会造成什么？ 当然是
index 的值相同，进而造成 mod、averageSize、startIndex、range 全部相同。那么最后 result.add(mqAll.get((startIndex + i) % mqAll.size())); 时，本来不同的 Consumer，会取到相同的 MessageQueue（举个例子，Consumer 1 和 Consumer 2 都取到了前 3 个 MessageQueue），从而造成有些 MessageQueue（如果有的话） 没有 Consumer 对其消费，而没有被消费，消息也在不停的投递进来，就会造成消息的大量堆积。
当然，现在的新版本从代码上看已经修复这个问题了，这个只是对之前的版本的原因做一个探索。
本篇文章已放到我的 Github github.com/sh-blog 中，欢迎 Star。微信搜索关注【SH的全栈笔记】，回复【队列】获取MQ学习资料，包含基础概念解析和RocketMQ详细的源码解析，持续更新中。 如果你觉得这篇文章对你有帮助，还麻烦点个赞，关个注，分个享，留个言。
我想为Heroku构建一个Rails3应用程序。他们使用Postgres作为他们的数据库，所以我通过MacPorts安装了postgres9.0。现在我需要一个postgresgem并且共识是出于性能原因你想要pggem。但是我对我得到的错误感到非常困惑当我尝试在rvm下通过geminstall安装pg时。我已经非常明确地指定了所有postgres目录的位置可以找到但仍然无法完成安装:$envARCHFLAGS='-archx86_64'geminstallpg--\--with-pg-config=/opt/local/var/db/postgresql90/defaultdb/po
尝试通过RVM将RubyGems升级到版本1.8.10并出现此错误:$rvmrubygemslatestRemovingoldRubygemsfiles...Installingrubygems-1.8.10forruby-1.9.2-p180...ERROR:Errorrunning'GEM_PATH="/Users/foo/.rvm/gems/ruby-1.9.2-p180:/Users/foo/.rvm/gems/ruby-1.9.2-p180@global:/Users/foo/.rvm/gems/ruby-1.9.2-p180:/Users/foo/.rvm/gems/rub
我的最终目标是安装当前版本的RubyonRails。我在OSXMountainLion上运行。到目前为止，这是我的过程:已安装的RVM$\curl-Lhttps://get.rvm.io|bash-sstable检查已知(我假设已批准)安装$rvmlistknown我看到当前的稳定版本可用[ruby-]2.0.0[-p247]输入命令安装$rvminstall2.0.0-p247注意:我也试过这些安装命令$rvminstallruby-2.0.0-p247$rvminstallruby=2.0.0-p247我很快就无处可去了。结果:$rvminstall2.0.0-p247Search
由于fast-stemmer的问题，我很难安装我想要的任何rubygem。我把我得到的错误放在下面。Buildingnativeextensions.Thiscouldtakeawhile...ERROR:Errorinstallingfast-stemmer:ERROR:Failedtobuildgemnativeextension./System/Library/Frameworks/Ruby.framework/Versions/2.0/usr/bin/rubyextconf.rbcreatingMakefilemake"DESTDIR="cleanmake"DESTDIR=
我有一个这样的哈希数组:[{:foo=>2,:date=>Sat,01Sep2014},{:foo2=>2,:date=>Sat,02Sep2014},{:foo3=>3,:date=>Sat,01Sep2014},{:foo4=>4,:date=>Sat,03Sep2014},{:foo5=>5,:date=>Sat,02Sep2014}]如果:date相同，我想合并哈希值。我对上面数组的期望是:[{:foo=>2,:foo3=>3,:date=>Sat,01Sep2014},{:foo2=>2,:foo5=>5:date=>Sat,02Sep2014},{:foo4=>4,:dat
我是rails的新手，想在form字段上应用验证。myviewsnew.html.erb.....模拟.rbclassSimulation{:in=>1..25,:message=>'Therowmustbebetween1and25'}end模拟Controller.rbclassSimulationsController我想检查模型类中row字段的整数范围，如果不在范围内则返回错误信息。我可以检查上面代码的范围，但无法返回错误消息提前致谢 最佳答案 关键是您使用的是模型表单，一种显示ActiveRecord模型实例属性的表单。c
当我尝试安装Ruby时遇到此错误。我试过查看this和this但无济于事➜~brewinstallrubyWarning:YouareusingOSX10.12.Wedonotprovidesupportforthispre-releaseversion.Youmayencounterbuildfailuresorotherbreakages.Pleasecreatepull-requestsinsteadoffilingissues.==>Installingdependenciesforruby:readline,libyaml,makedepend==>Installingrub
我正在尝试使用boilerpipe来自JRuby。我看过guide从JRuby调用Java，并成功地将它与另一个Java包一起使用，但无法弄清楚为什么同样的东西不能用于boilerpipe。我正在尝试基本上从JRuby中执行与此Java等效的操作:URLurl=newURL("http://www.example.com/some-location/index.html");Stringtext=ArticleExtractor.INSTANCE.getText(url);在JRuby中试过这个:require'java'url=java.net.URL.new("http://www
我意识到这可能是一个非常基本的问题，但我现在已经花了几天时间回过头来解决这个问题，但出于某种原因，Google就是没有帮助我。(我认为部分问题在于我是一个初学者，我不知道该问什么......)我也看过O'Reilly的RubyCookbook和RailsAPI，但我仍然停留在这个问题上.我找到了一些关于多态关系的信息，但它似乎不是我需要的(尽管如果我错了请告诉我)。我正在尝试调整MichaelHartl'stutorial创建一个包含用户、文章和评论的博客应用程序(不使用脚手架)。我希望评论既属于用户又属于文章。我的主要问题是:我不知道如何将当前文章的ID放入评论Controller。
我在rspec中收到来自webkit驱动程序的以下消息:Capybara::Driver::Webkit::WebkitInvalidResponseError:UnabletoloadURL:http://127.0.0.1:44923/posts几天前它成功了。问题出在save_page方法上。有什么问题吗？ 最佳答案 当我的页面出现错误时，我收到过类似的错误消息。您应该通过在测试模式下启动服务器(railss-etest)并自行访问页面来手动检查情况是否如此。 关于ruby-on-