ActiveMQ配置策略

1.消息发送

1.异步发送

消息生产者使用持久（persistent）传递模式发送消息的时候，Producer.send() 方法会被阻塞，直到 broker 发送一个确认消息给生产者，这个确认消息暗示生产者 broker已经成功地将它发送的消息路由到目标目的并把消息保存到二级存储中。这个过程通常称为同步发送。但有一个例外，当发送方法在一个事物上下文中时，被阻塞的是 commit 方法而不是 send 方法。commit 方法成功返回意味着所有的持久消息都以被写到二级存储中。

同步发送持久消息能够提供更好的可靠性，但这潜在地影响了程序的相应速度，因为在接受到 broker 的确认消息之前应用程序或线程会被阻塞。如果应用程序能够容忍一些消息的丢失，那么可以使用异步发送。异步发送不会在受到 broker 的确认之前一直阻塞 Producer.send 方法。如果想启动异步传送可以把 connector uri 的 jms.useAsyncSend 选项设为 true，如下所示：

tcp://localhost:61616?jms.useAsyncSend=true

从 ActiveMQ 5 开始可以控制异步发送流。也就是说，在受到 broker 的确认应答之前，生产者能够传送消息给 broker 的最大信息量。即使是异步发送消息，生产者也是在收到 broker 的确认应答后才把下一条消息传送给 broker。当使用异步传送的时候，可以设置jms.producerWindowSize（单位为字节）属性，当生产者中等待发送的信息量到达设置的值时，即使没有收到 broker 的应答消息，生产者同样会把这些消息发给 broker。如下面的示例设置：

tcp://localhost:61616?jms.useAsyncSend=true&jms.producerWindowSize=1024000

2. ActiveMQ讯息策略

3. 消息重发策略

1．在使用事务的Session中，调用rollback()方法；

2．在使用事务的Session中，调用commit()方法之前就关闭了Session;

3．在Session中使用CLIENT_ACKNOWLEDGE签收模式，并且调用了recover()方法。

4.消息持久化与非持久化

采用持久传输时，传输的消息会保存到磁盘中(messages
are persisted to disk/database)，即“存储转发”方式。先把消息存储到磁盘中，然后再将消息“转发”给订阅者。

采用非持久传输时，发送的消息不会存储到磁盘中。

采用持久传输时，当Borker宕机 恢复后，消息还在。采用非持久传输，Borker宕机重启后，消息丢失。比如，当生产者将消息投递给Broker后，Broker将该消息存储到磁盘中，在Broker将消息发送给Subscriber之前，Broker宕机了，如果采用持久传输，Broker重启后，从磁盘中读出消息再传递给Subscriber；如果采用非持久传输，这条消息就丢失了。

5.死信队列

DLQ-死信队列(Dead Letter Queue)用来保存处理失败或者过期的消息.缺省的死信队列是ActiveMQ.DLQ，如果没有特别指定，死信都会被发送到这个队列。默认情况下持久消息过期，会被送到DLQ，非持久消息不会送到DLQ

消息进入死信队列条件:

1.给ActiveMQConnectionFactory配上重发机制;

2.给DefaultMessageListenerContainer配置事务;

6. Consumer特性详解与优化

7.慢速消费者(Slow
Consumer)

慢速，是相对于producer而言；简单来说，producer不断产生新的消息，broker端在内存中已经积压的足够多(比如cacheLimit已满)，但是在转发给某个consumer时，发现此consumer仍然有大量的消息尚没有消费(ACK),broker会认为此consumer是慢速的。

在Queue中，如果已发送(dispatched)但没有消费(unAck)的消息条数 > prefetchSize时，此consumer被标记为Slow。

在Topic中，如果cacheLimit已满，但是需要向此订阅者发送的消息量 >
prefetchSize时，此订阅者被标记为Slow。

简单描述为： 快速的producer生产的消息，不能被消费者及时的消费，而导致在broker端积压。

解决方案:

1) 关闭Slow Consumer

brorker端一旦发现slow consumer，就将它注册到慢速消费者列表中，此后将有额外的线程扫描并关闭它们，其中abortConnection表示，是否关闭底层的transport，默认为false，此时将会通过transport向client端发送一个指令(其中包括consumerId)，当client端(Session)接收之后，将会调用consumer.close()方法；如果此值为true，将会导致底层的transport链接被关闭，这是很粗暴的办法，不过如果client端多个consumer共享一个connection的话，会导致所有的consumer被关闭，还是那句话：猪一样的队友，害了整个团队。

2) 抛弃旧消息(仅对Topic有效，仅对nondurable订阅者有效)

topic支持支持3种移除策略：

(1) oldestMessageEvictionStrategy表示移除最旧的消息

(2) uniquePropertyMessageEvictionStrategy表示移除根据属性值筛选消息并移除最旧的

(3)
OldestMessageWithLowestPriorityEvictionStrategy表示在旧消息中移除权重最低的。

3) 写入临时文件

       对于Queue而言，支持storeCursor，vmQueueCursor ， fileQueueCursor。其中storeCursor是一个“综合”策略，持久化消息使用fileQueueCurosr支持，非持久化消息使用vmQueueCursor支持。vmQueueCursor基于内存，fileQueueCursor表示将数据写入本地临时文件(由tempDataStore决定)。

对于Topic而言，我们也可以根据订阅者的类型，来决定如果处理那些滞留的非持久化消息。

4) offlineDurableSubscriberTimeout

对于durable订阅者，如果订阅者“离线”，那么Broker将会一直保存属于它的消息，因此消息也会以为它而不能删除，导致积压。通常，订阅者离线的时间是无法预估的，有可能此订阅者永远都不会再上线(可能因为durable订阅者本来应该cancel，但是开发者却忘记了)，这对broker来说是致命的。我们需要限制durable订阅者离线的时间，如果超过时间，那么订阅者将会被移除，消息也会因此而删除。

8. Activemq消息确认机制

optimizeAck表示是否开启“优化ACK”, 只有在为true的情况下，prefetchSize(下文中将会简写成prefetch)以及optimizeAcknowledgeTimeout参数才会有意义, "optimizeAcknowledgeTimeout"选项只能在brokerUrl中配置, 在destinationUri中指定prefetchSize(预获取)选项,
其中brokerUrl参数选项是全局的，如果同时指定，brokerUrl中的参数选项值将会被覆盖;

当consumer端使用MessageListener异步获取消息时: prefetch>=1

当consumer端使用receive()方法同步获取消息时: prefetch>=0

prefetch=0, receive()方法将会首先发送一个PULL指令并阻塞，直到broker端返回消息为止;

prefetch>0, broker端将会批量push给client 一定数量的消息(<=
prefetch),client端会把这些消息(unconsumedMessage)放入到本地的队列中;

原文地址：https://www.cnblogs.com/jinloooong/p/9609420.html