activemq消息积压处理（rabbitmq消息阻塞如何解决）

本文目录

• rabbitmq消息阻塞如何解决
• 关于ActiveMQ的配备怎么解决
• mq中如何保证消费者顺序消费

rabbitmq消息阻塞如何解决

按照JMS规范，为了保证可靠性，所有的消息都应该是发送到broker，然后交由broker来投递的。也即是说其实JMS是不建议或不支持传输文件的。
对于比较小的文件，简单的处理方式是先读取所有的文件成byte，然后使用ByteMessage，把文件数据发送到broker，像正常的message一样处理。对于大文件，例如1GB以上的文件，这么搞直接把client或是broker给oom掉了。
这种方式仅仅适用于小文件的传输。特别是如果broker端使用数据库作为存储，message序列化以后存放于blob字段，文件传输频繁或是稍微有点大，写入效率极低。
直接传输文件
为了解决传输大文件的问题，ActiveMQ在jms规范之外引入了jms streams的概念。PTP模式下，连到同一个destination的两端，可以通过broker中转来传输大文件。
发送端使用connection.createOutputStream打开一个输出流，往流里写文件。
OutputStream out =connection.createOutputStream(destination);
接收端则简单的使用connection.createInputStream拿到一个输入流，从中读取文件数据即可。
InputStream in = connection.createInputStream(destination)

关于ActiveMQ的配备怎么解决

关于ActiveMQ的配置
目前常用的消息队列组建无非就是MSMQ和ActiveMQ，至于他们的异同，这里不想做过多的比较。简单来说，MSMQ内置于微软操作系统之中，在部署上包含一个隐性条件：Server需要是微软操作系统。（对于这点我并去调研过MSMQ是否可以部署在非微软系统，比如:Linux，只是拍脑袋想了想，感觉上是不可以）。对于ActiveMQ，微软系统和Linux都是可以部署的。从功能方面来说，一般最常用的就是：消息的收/发，感觉差异不大。从性能上来说，一般的说法是ActiveMQ略高。在稳定性上，个人感觉MSMQ更好。如果这两种常用队列都用过的同学，应该来说最大的差异在于：MSMQ如果要访问远程队列（比如机器A上的程序访问机器B上的队列），会比较恶心。在数据量比较大的情况之下，一般来说队列服务器会专门的一台或者多台（多台的话，用程序去做热备+负载比较方便，也不需要额外的硬件成本。简单来说做法可以这样：消息发送的时候随机向着多台队列服务器上发送消息；接受的时候开多个线程去分别监听；热备方面，可以维护一个带状态的队列连接池，如果消息收发失败那么将状态置为不可用，然后起一个线程去定时监测坏的连接是否可用，这个过程一般情况下可以不用加锁，为什么，大家根据各自需要去取舍吧）。最近搞完了短彩信的网关连接服务，这两种队列我均使用了。大致的过程是这样的：上层应用如果要发端彩信，那么将消息直接发送至ActiveMQ（目前用的就是上面说的多台热备+负载，因为实际中下行量非常大5千万条/天以上），然后端彩信网关连接服务部署多套，每套均依赖本机的MSMQ。为什么呢？用ActiveMQ的原因是：上层应用程序和网关连接服务彼此独立，消息需要跨机访问。用MSMQ的原因是：ActiveMQ中的数据是一条不分省的大队列，网关连接服务需要按省流控，所以端彩信网关连接服务：首先把消息从ActiveMQ取出来，然后存至本机上的分省MSMQ，这样做另外的一个好处就是：ActiveMQ不至于过多挤压，他的数据会分摊到N台短彩信网关连接服务所部署的机器上的MSMQ之中，也就说MSMQ可以起到分摊数据和缓冲的作用。
在之前的随笔中，已经介绍过MSMQ，现在先介绍一下ActiveMQ一些配置，目前好像ActiveMQ配置上的介绍还比较少。以下是自己总结一些相关资料，贴出来给大家共享
一）问题分析和解决
1）KahaDb和AMQ Message Store两种持久方式如何选择？
官方：
From 5.3 onwards - we recommend you use KahaDB - which offers improved scalability and recoverability over the AMQ Message Store.
The AMQ Message Store which although faster than KahaDB - does not scales as well as KahaDB and recovery times take longer.
非官方：
kaha文件系统实际上上是一个文件索引系统，有两部分组成，一个是数据文件系统，由一个个独立的文件组成，缺省文件大小是32M大（可配置），另外一个是索引文件系统，记录消息在数据文件中的位置信息以及数据文件中的空闲块信息。数据文件是存储到硬盘上的，索引文件是缓存在内存中的。所以这个存储系统对大消息存储有利，象我们的memberId之类的文本消息，实际上是浪费，索引比消息还大，哈。
我方分析：
推荐： Amq持久方式
理由：虽然官方推荐使用KahaDB持久方式，但其提到的优势：可伸缩性和恢复性较好，对于我们实际的应用意义不大。从我们自己的使用经验来看，KahaDB持久方式，Data文件是一个大文件（感觉文件过大后，造成队列服务瘫死的可能性会增大），从官网的相关配置（附录1）也找不到哪里可以设置数据的文件的最大Size。）而Amq持久方式可以设置Data文件最大Size，这样可以保证即时消息积压很多，Data文件也不至于过大。
2）错误：Channel was inactive for too long
解决方法：
在建立连接的Uri中加入： wireFormat.maxInactivityDuration=0
参考资源：
***隐藏网址***
You can do the following to fix the issues:
1) Append max inactivity duration to your Uri in the format below: wireFormat.maxInactivityDuration=0
2) Use the same Uri at the client side as well as at the server side
Regards,
如果不这样设置，对应的错误会出现：
2008-05-07 09:22:56,343 Async exception with no exception listener: org.apache.activemq.transport.InactivityIOException: Channel was inactive for too long: localhost/127.0.0.1:61616
org.apache.activemq.transport.InactivityIOException: Channel was inactive for too long: localhost/127.0.0.1:61616
ActiveMQ的tcp url：tcp://localhost:61616后面要加入?wireFormat.maxInactivityDuration=0 这样的参数，否则当一段时间没有消息发送时会抛出 "Channel was inactive for too long"异常
3）错误：Wire format negotiation timeout: peer did not send his wire format.
解决方法：
1）关闭ActiveMqLog4j
打开：conf/log4j.properties
将：log4j.rootLogger=INFO, console, logfile
修改为：log4j.rootLogger=OFF
2）在建立连接的Uri中加入： maxInactivityDurationInitalDelay=30000
例如北京的测试环境连接Uri：
tcp://192.168.199.80:61616?wireFormat.maxInactivityDuration=0&maxInactivityDurationInitalDelay=30000&connection.AsyncSend=true
参考资源：
***隐藏网址***
If you get exception like this,it can mean one of three things:
1. You’re connecting to the port not used by ActiveMQ TCP transport
Make sure to check that you’re connecting to the appropriate host:port
2. You’re using log4j JMS appender and doesn’t filter out ActiveMQ log messages
Be sure to read How do I use log4j JMS appender with ActiveMQ and more importantly to never send ActiveMQ log messages to JMS appender
3. Your broker is probably under heavy load (or network connection is unreliable), so connection setup cannot be completed in a reasonable time
If you experience sporadic exceptions like this, the best solution is to use failover transport, so that your clients can try connecting again if the first attempt fails. If you’re getting these kind of exceptions more frequently you can also try extending wire format negotiation period (default 10 sec). You can do that by using wireFormat.maxInactivityDurationInitalDelay property on the connection URL in your client.
For example
tcp://localhost:61616?wireFormat.maxInactivityDurationInitalDelay=30000
will use 30 sec timeout.（貌似有问题！！！）
4）错误：Out of memory
解决方法：
1） 设置Java最大内存限制为合适大小：
Bin/activemq.bat 中ACTIVEMQ_OPTS=-Xmx512M（默认是512）
2）Activemq.xml配置节：systemUsage/ systemUsage配置大小合适，并且特别注意：大于所有durable desitination设置的memoryUsage之和。
备注：
1）尖括号：“》”代表通配符
2）ACTIVEMQ_OPTS的配置〉=memoryUsage中配置〉=所有durable desitination设置之和
3）SystemUsage配置设置了一些系统内存和硬盘容量，当系统消耗超过这些容量设置时，amq会“slow down producer”，还是很重要的。
参考资料：
***隐藏网址***
***隐藏网址***
对于MQ的内容实用是可管理和可配置的。首先需要判断的是MQ的哪部分系统因内存不足而导致泄漏，是JVM，broker还是消费者、生产者？
一、内存管理
JVM内存管理：
1. 用bin/activemq命令在独立JVM中运行broker。用-Xmx和-Xss命令即可（activemq.bat文件中修改ACTIVEMQ_OPTS选项参数即可）；
2. 默认情况下，MQ用512M的JVM；
broker内存管理：
1. broker使用的内存并不是由JVM的内存决定的。虽然受到JVM的限制，但broker确实独立管理器内存；
2. systemUsage和destination的内存限制与broker内存息息相关；
3. MQ中内存的关系是：JVM-》Broker-》broker features;
4. 所有destination的内存总量不能超过broker的总内存；
消费者：
1. 由于消息大小可以配置，prefetch limit往往是导致内存溢出的主要原因；
2. 减少prefetch limit的大小，会减少消费者内存中存储的消息数量；
生产者：
1. 除非消息数量超过了broker资源的限制，否则生产者不会导致内存溢出；
2. 当内存溢出后，生产者会收到broker的阻塞信息提示；
二、其他
将消息缓冲之硬盘：
1. 只有当消息在内存中存储时，才允许消息的快速匹配与分发，而当消费者很慢或者离开时，内存可能会耗尽；
2. 当destination到达它的内存临界值时，broker会用消息游标来缓存非持久化的消息到硬盘。
3. 临界值在broker中通过memoryUsage和systemUsage两个属性配置，请参考activemq.xml；
4. 对于缓慢的消费者，当尚未耗尽内存或者转变为生产者并发控制模式前，这个特性允许生产者继续发送消息到broker；
5. 当有多个destination的时候，默认的内存临界值可能被打破，而这种情况将消息缓存到硬盘就显得很有意义；
6. precentUsage配置：使用百分比来控制内存使用情况；
多个线程：
1. 默认情况下，MQ每个destination都对应唯一的线程；
2. -Dorg.apache.activema.UseDedicatedTaskRunner=false（activemq.bat文件中修改ACTIVEMQ_OPTS选项参数即可），用线程池来限制线程的数量，从而减少内存消耗;
大数据传输：
1. destination policies--maxPageSize：控制进入内存中的消息数量；lazyDispatch:增加控制使用当前消费者列表的预取值；
2. 使用blogMessage或者streamsMessage类型来进行大量文件的传输；
泄漏JMS资源：
1. 当session或者producer或者consumer大量存在而没有关闭的时候；
2. 使用PooledConnectionFactory；

mq中如何保证消费者顺序消费

通常mq可以保证先到队列的消息按照顺序分发给消费者消费来保证顺序，但是一个队列有多个消费者消费的时候，那将失去这个保证，因为这些消息被多个线程并发的消费。但是有的时候消息按照顺序处理是很重要的，那我们该如何来保证消息的顺序呢，下面将从activemq和rocketmq来看看，它们是如何来保证消息的顺序问题的？我们还可以有别的处理方案么？

1、利用Activemq的高级特性：consumer之独有消费者（exclusive consumer）

在ActiveMQ4.x中可以采用Exclusive Consumer，broker会从queue中，一次发送消息给一个消费者，这样就避免了多个消费者并发消费的问题，从而保证顺序，配置如下：

.当在接收信息的时候有一个或者多个备份接收消息者和一个独占消息者的同时接收时候，无论两者创建先后，在接收的时候，均为独占消息者接收。
.当在接收信息的时候，有多个独占消费者的时候，只有一个独占消费者可以接收到消息。
.当有多个备份消息者和多个独占消费者的时候，当所有的独占消费者均close的时候，只有一个备份消费者接到到消息。
.当主消费者挂了话，会立刻启用故障切换转移到下一台消费者继续消费

独占消息就是在有多个消费者同时消费一个queue时，可以保证只有一个消费者可以消费消息，这样虽然保证了消息的顺序问题，不过也带来了一个问题，就是这个queue的所有消息将只会在这一个主消费者上消费，其他消费者将闲置，达不到负载均衡分配，而实际业务我们可能更多的是这样的场景，比如一个订单会发出一组顺序消息，我们只要求这一组消息是顺序消费的，而订单与订单之间又是可以并行消费的，不需要顺序，因为顺序也没有任何意义，有没有办法做到呢？答案是可以的，下面就来看看activemq的另一个高级特性之messageGroup。
2、利用Activemq的高级特性：messageGroups

Message Groups特性是一种负载均衡的机制。在一个消息被分发到consumer之前，broker首先检查消息JMSXGroupID属性。如果存在，那么broker会检查是否有某个consumer拥有这个message group。如果没有，那么broker会选择一个consumer，并将它关联到这个message group。此后，这个consumer会接收这个message group的所有消息，直到：

Consumer被关闭
Message group被关闭，通过发送一个消息，并设置这个消息的JMSXGroupSeq为-1

如上图所示，同一个queue中，拥有相同JMSXGroupID的消息将发往同一个消费者，解决顺序问题，不同分组的消息又能被其他消费者并行消费，解决负载均衡的问题，两全其美啦！

Rocketmq处理方案

那rocketmq又是如何保证消息顺序消费的问题呢？

Rocketmq跟传统的MQ有一点区别，这里有必要讲一下topic的概念，Topic是RocketMQ中的一个重要概念，RocketMQ的各组件都是围绕着Topic建立起对应关系的，在RocketMQ官方文档和本文中， Topic在不同的语境下被赋予了两种不同的语义：

1)消息的Topic属性值：在描述Consumer的订阅设置信息或消息的属性时。

2)Topic属性为某个值的消息(单个消息或消息集合)：在描述Broker，Producer和Consumer的对应关系，Queue以及负载均衡策略时。

topic和queue的对应关系是一个topic拥有多个queue，当producer往broken发送消息时，消息会存储在topic下的不同队列中，而一个队列只会被一个consumer消费，这样消息户被均衡负载到不同的队列下，也就是会被多个消费者并行消费，顺序就无法保证了。该怎么办呢？答案是把需要顺序消费的消息发送到同一台broker server下的同一个队列，而这些消息也只会被同一个消费者消费，这样就可以保证严格的顺序了，如下图：

1、消息要有顺序，首先得保证producer发送消息有顺序，如上图msg1，msg2，msg3发送到queue0是要有顺序的，只要producer等待前面的消息发送成功，在发送后面的消息就完全可以保证了，

2、假设msg1发送给consumer1，消费没有响应，该怎么办呢？是继续发送msg2还是重新发送msg1，一般为了保证消息一定被消费，肯定会选择重发msg1到下一台消费者consumer2。

3、消费端1没有响应Server时有两种情况，一种是msg1确实没有到达(数据在网络传送中丢失)，另外一种消费端已经消费msg2且已经发送响应消息，只是MQ Server端没有收到。如果是第二种情况，重发msg1，就会造成msg1被重复消费。也就引入了消息重复问题，那就要幂等了。

Rocketmq同样做到了保证消息的顺序情况下，均衡消费的消费消息。

综上，我看到，在分布式系统中，要想消息有顺序的被消费，无论是Activemq还是Rocketmq都要想办法让有顺序的消息被同一消费者消费，而不是并发的消费，在消费者消费成功后，接着才会消费下一个消息，这样就可以保证了严格的顺序。