pzxsheng

有种相见不敢见的伤痛,有种爱还埋藏在心中

activeMQ指南针_Queue完整分析

原文地址:http://www.360doc.com/content/13/0315/16/11765546_271697514.shtml

在接触activeMQ的这一段时间里,我们还是保持开始对它的态度,它是个优秀的开源消息中间件。消息中间件是个非常重要的搭建企业应用系统的重要组件,我们在不断深入分析activeMQ的过程中,发现直到5.1这个版本,都还是存在不少问题,有些是很致命,但正因为如此,我们更加坚定了要全面掌握activeMQ,我们不想重新做“轮子”,但我们要具备在轮子坏了或不好用的情况下,要能独立解决碰到的这些问题。下面我们通过分析网友提出的一个典型的问题场景,来作为我们指南针计划的结束。

Queue作为activeMQ里面一个很重要的通讯方式,网友的场景如下:

测试queue持久化消息时,发送接收20W条消息。打开消息消费者,连上再断开,反复进行这步操作,能接收到消息,接收端有时候会阻塞,但不能完全接收完20W条消息。(其实5000条就会发生问题,不用20W这么多)

       相关背景知识:

       因为这是5.1版本的一个非常严重的bug,所以我们会比较详细的进行分析。(我们在最终解决问题后,上activeMQ官网上发现它最新的源码是解决了该问题的,但这并不影响这个问题的典型性)。下面我们将从3个方面来分析:Queue消息的接收和发送、内存使用机制、消息的审查(audit)、消息在文件中的存储机制。

l         Queue消息的接收和发送



 

Queue接收消息并发给需要的消费者,具体过程如下:

1.  Queue从消息生产者接收消息。

2.  Queue使用一个“存储指针”来接收这些消息。当内存有空闲区域时,“存储指针”把消息放到内存中,当内存不够时,则把消息们存入磁盘文件。

3.  当有活动的(active)的消息消费者时,Queue会首先把“存储指针”的内存中的消息送给消费者,当内存的消息被消费掉,则从磁盘文件中再读入其他的消息(出问题处),直至消息都被消费掉了。

其中最关键的方法是Queue类里的doPageIn()

 

l         内存使用机制

activeMQ为了适应企业级的365*24的使用,在内存使用方面非常慎重,任何消息只有在内存里有空闲区域时,才能放到内存里,之后才能发给消费者。当消息被消费者消耗掉了后,确认信息会发给activeMQQueue接收到这些确认消息后,会把那些被确认的消息所占用的内存释放掉。

 

l         消息的审查(audit)

为了防止消息的重复发送,activeMQ采用了一个审查机制,它负责审查某条消息是否重复。它是一个最近最久未使用算法(LRU)队列。每个队列元素它是一个bit数组,它的运行机制如下所示:



 

 

       消息是一个个按照顺序进入bit数组,具体算法answer = (index - firstIndex) / BitArray.LONG_SIZE,其中:

BitArray.LONG_SIZE是每个bit数组的大小。

Index是消息的编号。(它是按照+1顺序增加的)

firstIndex是整个LRU队列的首Index,这个值会经常变化,因为当达到LRU的上限时,老的一批就被清除了,firstIndex += BitArray.LONG_SIZE(出问题处)

 

l         消息在文件中的存储机制

存放在文件中的消息,它们是按照如下方式进行组织的:



 

每个消息都知道它的上一个和下一个消息,当它自身被删除后,相应的关系会进行调整。

 

问题原因分析:

    因为activeMQ在编码实现的时候,原本的想法应该是这样的:

1.  从生产者接收消息,如果Queue有可用的内存就放在内存中,没有则存入文件中。

2.  Queue发送消息给消费者时,先发送已经保存在内存中的消息。

3.  当内存中消息发送完后,顺序读入(这里是关键)文件中的消息,通过消息的审查机制,确认不是重复消息,则放入内存中供后续操作使用。

但是activeMQ5.1版本的实现,问题就出在第三步的顺序读入。因为从文件中读入它有个先决条件,那就是必须要有可用的内存,如果没有可用的话,就放弃本次消息读入,并且应该放弃这次读取操作。但是5.1版本是继续往下读,这就导致顺序错乱,使得当内存可用的时候,读入的消息在进行审查的时候,发生错误,错误认为它们是重复消息。这就导致发送20W条消息,不能保证完全收到。

 

解决方案:

KahaReferenceStore的方法recoverNextMessages里的

if (entry != null) {

                int count = 0;

                do {

                    ReferenceRecord msg = messageContainer.getValue(entry);

                    if (msg != null ) {

                        if ( recoverReference(listener, msg)) {

                            count++;

                            lastBatchId = msg.getMessageId();

                        }

                    } else {

                        lastBatchId = null;

                    }

                    batchEntry = entry;

                    entry = messageContainer.getNext(entry);

                } while (entry != null && count < maxReturned && listener.hasSpace());

            }

 

改为

         if (entry != null) {

                int count = 0;

                do {

                    ReferenceRecord msg = messageContainer.getValue(entry);

                    testTheNextMsgId(msg.getMessageId().toString());

                    if (msg != null )

                    {

                        if ( recoverReference(listener, msg))

                        {

                            count++;

                            lastBatchId = msg.getMessageId();

                            batchEntry = entry;

                            entry = messageContainer.getNext(entry);

                        }

                        else

                        {

                           break;

                        }

                    }

                    else

                    {

                        lastBatchId = null;

                        batchEntry = entry;

                        entry = messageContainer.getNext(entry);

                    }

                } while (entry != null && count < maxReturned && listener.hasSpace());

            }

 

 

activeMQ指南针计划的结束,但它又是个新开始,我们通过这个计划收获了我们想要的东西了,同时我们不仅为各位朋友答疑解疑,也提供了activemqSpanner这个工具作为消息网络拓扑图工具。再一次感谢各位朋友对我们的信任。

        现在,我们正式启动activeMQ笑脸计划。它的目的不再是给大家提供解决问题的方向,而是直接解决大家碰到的各种问题,给大家带去笑脸。它将是一个长期坚持的事情,任何关于activeMQ使用过程的疑惑、问题、bug、功能改进,都可以在这个计划里交流。所有在笑脸计划中提出的问题、功能改进、解决方案,都将完全通过网络无偿分享给所有人。

posted on 2013-03-15 16:24 科菱财神 阅读(694) 评论(0)  编辑  收藏 所属分类: ActiveMQ


只有注册用户登录后才能发表评论。


网站导航:
 

导航

<2013年3月>
242526272812
3456789
10111213141516
17181920212223
24252627282930
31123456

统计

常用链接

留言簿(1)

随笔分类

随笔档案

搜索

最新评论

阅读排行榜

评论排行榜