ActiveMQ学习笔记-分发策略

偏执的太偏执、 2022-06-16 12:48 268阅读 0赞

## 异步发送 ##

ActiveMQ默认的发送模式是异步发送，如果我们使用场景是非事务类型或者需要持久化消息，允许少量的消息丢失的话，推荐使用异步发送。同步发送需要增加ack的确认，这样子会增大时延和系统消耗。

cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.useAsyncSend=true");

((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);

((ActiveMQConnection)connection).setUseAsyncSend(true);

## 分发策略 ##

可插拔的分发策略仅仅能使用在话题上，队列的消息分发策略比较固定，轮询（默认）或者粘性顺序，同时我们也应该了解prefetch的数值的意义[prefetch][]。  
默认的prefetch值是十分大的，默认的分发策略总是尽可能的填满prefetch的缓冲区。  
这里存在很多的用例且有时候默认的配置并不是很好的工作；比如你发送了很少数量的消息，但是broker要等到一定大数量的消息才发往一个消费者，同时每个消息的处理都需要消耗一定的时间，那么大量的消息将会增加处理的时间。

strictOrderDispatch表示在直到当前消费者的prefetch缓冲区满了之后才选择下一个消费者进行消息的分发。

<policyEntry queue=">" strictOrderDispatch="false" />

### 话题的分发策略 ###

org.apache.activemq.broker.region.policy.DispatchPolicy类的实现是话题分发的具体策略实现。默认的分发策略是org.apache.activemq.broker.region.policy.SimpleDispatchPolicy，而network 中的分发策略是将消息往优先级高的进行发送org.apache.activemq.broker.region.policy.PriorityNetworkDispatchPolicy 。

## 消息游标 ##

使用ActiveMQ会遇到的一个共同的问题就是非持久化消息会耗尽内存缓冲区。  
ActiveMQ5.0.0开始，新的内存模型允许我们将消息缓存在磁盘存储页面中，同时也可以直接将消息从生产者传递到消费者（在消息已经持久化后）。  
![Dispatching Message for Fast Consumers][]

![Dispatching Message if Dispatch Queue is Full][]

### 游标类型 ###

ActiveMQ 5.0的有两种游标可以选择

*  VM Cursor
 *  File based Cursor

**VM Cursor**  
速度快，但是在处理慢消费者的时候存在缺点  
![VM Cursor][]

**File based Cursor**  
File based Cursor是从VM Cursor中派生出来的，当broker达到内存限制的时候，能够将消息分页存储到磁盘中，适用消息存储十分慢但是消费处理十分快的场景。  
![File based Cursor][]

**Paging for Non-Persistent Messages**  
非持久化消息直接把消息传送游标去。  
![Paging for Non-Persistent Messages][]

**配置游标**  
默认使用存储游标，可以为不同的队列配置不同的游标策略。

话题订阅

有效的订阅类型是vmCursor，fileCursor，有效的持久化订阅游标类型是storeDurableSubscriberCursor，storeDurableSubscriberCursor和fileDurableSubscriberCursor，默认是storeDurableSubscriberCursor。

<destinationPolicy>
          <policyMap>
            <policyEntries>
              <policyEntry topic="org.apache.>" producerFlowControl="false" memoryLimit="1mb">
                <dispatchPolicy>
                  <strictOrderDispatchPolicy />
                </dispatchPolicy>
                <deadLetterStrategy>
                  <individualDeadLetterStrategy  topicPrefix="Test.DLQ." />
                </deadLetterStrategy>
                <pendingSubscriberPolicy>
                    <vmCursor />
                </pendingSubscriberPolicy>
                <pendingDurableSubscriberPolicy>
                    <vmDurableCursor/>
                </pendingDurableSubscriberPolicy>
              </policyEntry>
            </policyEntries>
          </policyMap>
    </destinationPolicy>

队列的游标类型有 storeCursor, vmQueueCursor 和 fileQueueCursor，默认是storeCursor。

## 优化确认 ##

ActiveMQ支持确认一个范围内的消息在单个批量操作，这个特性默认是关闭的，开启这个特性会减少broker的负载，可以提升某些情况下系统的吞吐量。

cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.optimizeAcknowledge=true");

((ActiveMQConnectionFactory)connectionFactory).setOptimizeAcknowledge(true);

((ActiveMQConnection)connection).setOptimizeAcknowledge(true);

setOptimizeAcknowledgeTimeOut用于设置优化确认的超时时间，默认是300毫秒，0表示不启用该特性。

## 生产者流量控制 ##

ActiveMQ4.0生产者的流量控制使用TCP的流量控制，这个策略是十分有效的但是在多个生产者和消费者之间共享同一个connection可能会导致死锁。  
ActiveMQ5.0以来，我们可以为每一个生产者配置共享的connection而无需suspend整个connection，流量控制意味着当broker检测到内存达到极限，或者文件存储满了，能够让消息的产生慢下来，阻塞生产者知道资源可用，或者接收到JMSException，这些信息我们是配置在节点systemUsage中。

值得注意的是默认的systemUsage配置在memoryLimit或者是systemUsage 达到极限时候是会阻塞生产者的，有时候这会让人误解为生产者被挂起了，实际上消费者只是在等待资源可用。

*  同步发送消息将会自动启用流量控制，除非你开启了useAsyncSend特性
 *  在开启了useAsyncSend特性后，内存达到限制后我们将不会接收到任何通知，我们需要配置ProducerWindowSize 用于在内存达到限制后启用流量控制

ProducerWindowSize是生产者在接收到ack之前往broker发送最大的字节数限制

ActiveMQConnectionFactory connctionFactory = ...
    connctionFactory.setProducerWindowSize(1024000);

或者我们在broker中的内存达到限制的时候通知到，我们可以配置alwaysSyncSend特性。

<destinationPolicy>
      <policyMap>
        <policyEntries>
          <policyEntry topic="FOO.>" producerFlowControl="false"/>
        </policyEntries>
      </policyMap>
    </destinationPolicy>

### Producer Flow Control 如何工作 ###

通过往生产者发送ack通知生产者当前的windows 大小的数量已经处理，可以继续发送下一个windowSize。

**优势**  
一个好的生产者将会等到返回ack后再继续发送消息从而避免流量泛滥。

**Configure Client-Side Exceptions**  
当broker资源不足时候，将send（）的阻塞操作替代微抛出异常也是选项之一。将sendFailIfNoSpace属性配置为true，那么broker将会抛出javax.jms.ResourceAllocationException异常，并传播到客户端。  
这样子send操作就不会被挂住，而是捕获到异常后再继续重试。下面配置的单位是毫秒

## 时序 ##

有时候保持话题消费者的消费顺序是很有必要的，broker可以保证同一个生产者发送的消息的顺序，但是对于多线程和异步处理，消息从不同到的生产者达到不同消费者的顺序是不同的  
比如生产P和Q，发送消息P1，P2，P3，和Q发送Q1，Q2，消费者可能接收到的消息是

consumer1: P1 P2 Q1 P3 Q2
    consumer2: P1 Q1 Q2 P2 P3

total Orrdering保证了每一个消费所见的消息的顺序都是同样的，但是这样子可能会带来性能上的消耗，但是对于事务类型系统这还是很重要的。

consumer1: P1 P2 Q1 P3 Q2
    consumer2: P1 P2 Q1 P3 Q2

[prefetch]: http://activemq.apache.org/what-is-the-prefetch-limit-for.html
[Dispatching Message for Fast Consumers]: /images/20220616/c5ff71c2cea44f3e918742779b6b8946.png
[Dispatching Message if Dispatch Queue is Full]: /images/20220616/5265ddf1c11644418c04397bba52c86f.png
[VM Cursor]: /images/20220616/5167b7a4c2e8453cafbaf2bc280272d5.png
[File based Cursor]: /images/20220616/315869f1e8734ee890e308d110842fc3.png
[Paging for Non-Persistent Messages]: /images/20220616/336f9b570d29466c8d9cba8a28c81045.png