`

ActiveMq生产者流量控制(Producer Flow Control)

 
阅读更多
<style type="text/css"> <!-- p {text-indent:2em} --> </style>

最近学习ActiveMq,昨日查看其配置文件activemq.xml的时候,被一行注释引到了http://activemq.apache.org/producer-flow-control.html页面,感觉挺有用,遂翻译之保存成文。作为自己的第一篇译作,必有诸多不妥之处,故附上原文作对照,欢迎拍砖。


Producer Flow Control生产者流量控制

In ActiveMQ 4.x flow control was implemented using TCP flow control. The underlying network connection of throttled consumers was suspended to enforce flow control limits. This strategy is very efficient but can lead to deadlocks if there are multiple producers and consumers sharing the same connection.

在ActiveMQ 4.x版本中,流量控制是用TCP流量控制实现的。受节制的消费者底层的网络连接被挂起,以强制进行流量控制限制。这个策略非常高效,但是如果有多个生产者和消费者共享同一个连接的时候,可能会导致死锁。

As of ActiveMQ 5.0, we can now individually flow control each producer on a shared connection without having to suspend the entire connection. By 'flow control' we mean that if the broker detects that the memory limit for the destination, or the temp- or file-store limits for the broker, have been exceeded, then the flow of messages can be slowed down. The producer will be either blocked until resources are availableorwill receive a JMSException: this behaviour is configurable and described in the section below on<systemUsage>.

在ActiveMQ5.0版本中,我们可以分别对一个共享连接上的各个生产者进行流量控制,而不需要挂起整个连接。“流量控制”意味着当代理(broker)检测到目标(destination)的内存,或temp-或file-store超过了限制,消息的流量可以被减慢。生产者将会被阻塞直至资源可用,或者收到一个JMSException异常:这种行为是可配置的,下面的<systemUsage>章节会描述到。

It's worth noting that the default<systemUsage>settings will cause the producer toblockwhen thememoryLimitor<systemUsage>limits are reached: this blocking behaviour is sometimes misinterpreted as a 'hung producer', when in fact the producer is simply diligently waiting until space is available.

值得注意的是,当内存限制或<systemUsage>限制达到的时候,<systemUsage>默认的设置会引起生产值阻塞:这种阻塞行为有时会被误解为“挂起的生产者”,而事实是生产者只是勤奋地等着空间可用。

  • Messages that are sent synchronously will automatically use per producer flow control; this applies generally to persistent messages which are sent synchronouslyunlessyou enable theuseAsyncSendflag.
  • 同步发送的消息将会自动对每一个生产者使用流量控制;除非你使能了useAsyncSend标志,否则这将对同步发送的持久化消息都适用。
  • Producers that useAsync Sends- generally speaking, producers of non-persistent messages - don't bother waiting for any acknowledgement from the broker; so, if a memory limit has been exceeded, you willnotget notfied. If you do want to be aware of broker limits being exceeded, you will need to configure the ProducerWindowSize connection option so that even async messages are flow controlled per producer.
  • 使用异步发送的生产者 ——一般来说,就是发送非持久化消息的生产者 —— 不需要等候来自代理的任何确认消息;所以,如果内存限制被超过了,你不会被通知。如果你真的想什么时候代理的限制被超过了,你需要配置ProducerWindowSize这一连接选项,这样就算是异步消息也会对每一个生产者进行流量控制。
ActiveMQConnectionFactory connctionFactory = ...
connctionFactory.setProducerWindowSize(1024000);

The ProducerWindowSize is the maximum number of bytes of data that a producer will transmit to a broker before waiting for acknowledgment messages from the broker that it has accepted the previously sent messages.

ProducerWindowSize是一个生产者在等到确认消息之前,可以发送给代理的数据的最大byte数,这个确认消息用来告诉生产者,代理已经收到先前发送的消息了。

Alternatively, if you're sending non-persisted messages (which are by default sent async), and want to be informed if the queue or topic's memory limit has been breached, then you can simply configure the connection factory to 'alwaysSyncSend'. While this is going to be slower, it will ensure that your message producer is informed immediately of memory issues.

或者,如果你要发送非持久化的消息(该消息默认是异步发送的),并且想要得到队列或者主题的内存限制是否达到,你只需将连接工厂配置为“alwaysSyncSend”。虽然这样会变得稍微慢一点,但是这将保证当出现内存问题时,你的消息生产者能够及时得到通知。

If you like, you can disable flow control for specific JMS queues and topics on the broker by setting theproducerFlowControlflag to false on the appropriate destination policy in the Broker configuration - e.g.

如果你喜欢,你可以通过在代理的配置中,将适当的目的地(destination)的策略(policy)中的producerFlowControl标志设置为false,使代理上特定的JMS队列和主题无效,例如:

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

seeBroker Configuration.

详情请看代理设置部分。

Note that, since the introduction of the new file cursor in ActiveMQ 5.x, non-persisted messages are shunted into the temporary file store to reduce the amount of memory used for non-persistent messaging. As a result, you may find that a queue's memoryLimit is never reached, as the cursor doesn't use very much memory. If you really do want to keep all your non-persistent messages in memory, and stop producers when the limit is reached, you should configure the<vmQueueCursor>.

注意,自从ActiveMQ 5.x中引入新的文件游标之后,非持久化消息被分流到了临时文件存储中,以此来减少非持久化消息传送使用的内存总量。结果就是,你可能会发现一个队列的内存限制永远达不到,因为游标不需要使用太多的内存。如果你真的想把所有的非持久化消息存放在内存中,并在达到内存限制的时候停掉生产者,你需要配置<vmQueueCursor>。

<policyEntry queue=">" producerFlowControl="true" memoryLimit="1mb">    
  <pendingQueuePolicy>
    <vmQueueCursor/>
  </pendingQueuePolicy>
</policyEntry>

The fragment above will ensure that all non-persistent queue messages are kept in memory, with each queue having a limit of 1Mb.

上面的片段可以保证所有的非持久化队列消息都保存在内存中,每一个队列的内存限制为1Mb。

How Producer Flow Control works生产者流量控制是如何工作的

If you are sending a persistent message (so that a response of theOpenWireMessage is expected then the broker will send the producer aProducerAckmessage. This informs the producer that the previous sending window has been processed, so that it can now send another window. Its kinda like consumer acks but in reverse.

如果你发送一条持久化消息(这样就会有一个对OpenWire响应消息的期望),代理将会给生产者发送一个应答消息。该消息会通知生产者其先前的发送窗口已经被处理了,所以它现在可以发送另外的窗口了。这有点像消费者应答,不过是反向的。

Advantage 优势

So a nice producer might wait for a producer ack before sending more data, to avoid flooding the broker (and forcing the broker to block the entire connection if a slow consumer occurs). To see how this works in source code, check out theActiveMQMessageProducercode.

所以,一个友好的生产者可以再发送更多的数据之前,等待生产者应答,以此来避免对代理的冲击(并且如果出现了一个比较慢的消费者,强制代理阻塞整个连接)。如果你想知道这部分的源代码是怎么实现的,可以看一下ActiveMQMessageProducer的代码。

Though a client can ignore the producer ACKs altogether and the broker should just stall the transport if it has to for slow consumer handling; though this does mean it'll stall the entire connection.

然而一个客户端可以完全忽略生产者的应答消息,并且处理慢消费者的时候,代理可以在需要的时候拖延传送;虽然这意味着它将拖延整个连接。

Configure Client-Side Exceptions配置客户端的异常

An alternative to the indefinite blocking of thesend()operation when no space is free on the broker is to instead configure that an exception to be thrown on the client-side. By configuring thesendFailIfNoSpaceproperty totrue, the broker will cause thesend()operation to fail with ajavax.jms.ResourceAllocationException, which will propagate to the client. Below is an example of this configuration:

应对代理空间不足,而导致不确定的阻塞 send()操作的一种替代方案,就是将其配置成客户端抛出的一个异常。通过将sendFailIfNoSpace属性设置为true,代理将会引起send()方法失败,并抛出javax.jms.ResourceAllocationException异常,传播到客户端。下面是一个配置的示例:

<systemUsage>
 <systemUsage sendFailIfNoSpace="true">
   <memoryUsage>
     <memoryUsage limit="20 mb"/>
   </memoryUsage>
 </systemUsage>
</systemUsage>

The advantage of this property is that the client can catch thejavax.jms.ResourceAllocationException, wait a bit and retry thesend()operation instead of just hanging indefinitely.

这个属性的好处是,客户端可以捕获javax.jms.ResourceAllocationException异常,稍等一下,并重试send()操作,而不是无限期地傻等下去。

Starting in version 5.3.1 thesendFailIfNoSpaceAfterTimeoutproperty has been added. This property causes thesend()operation to fail with an exception on the client-side, but only after waiting the given amount of time. If space on the broker is still not freed after the configured amount of time, only then does thesend()operation fail with an exception to the client-side. Below is an example:

从5.3.1版本之后,sendFailIfNoSpaceAfterTimeout属性被加了进来。这个属性同样导致send()方法失败,并在客户端抛出异常,但仅当等待了指定时间之后才触发。如果在配置的等待时间过去之后,代理上的空间仍然没有被释放,仅当这个时候send()方法才会失败,并且在客户端抛出异常。下面是一个示例:

<systemUsage>
 <systemUsage sendFailIfNoSpaceAfterTimeout="3000">
   <memoryUsage>
     <memoryUsage limit="20 mb"/>
   </memoryUsage>
 </systemUsage>
</systemUsage>

The timeout is defined in milliseconds so the example above waits for three seconds before failing thesend()operation with an exception to the client-side. The advantage of this property is that it will block for the configured amount of time instead of failing immediately or blocking indefinitely. This property offers not only an improvement on the broker-side, but also an improvement for the client so it can catch the exception, wait a bit and retry thesend()operation.

定义超时的单位是毫秒,所以上面的例子将会在使send()方法失败并对客户端抛出异常之前,等待三秒。这个属性的优点是,它仅仅阻塞配置指定的时间,而不是立即另发送失败,或者无限期阻塞。这个属性不仅在代理端提供了一个改进,还对客户端提供了一个改进,使得客户端能捕获异常,等待一下并重试send()操作。

Disabling Flow Control使流量控制无效

A common requirement is to disable flow control so that message dispatching continues until all available disk is used up by pending messages (whether persistent or non persistent messaging is configured). To do this enableMessage Cursors.

一个普遍的需求是使流量控制无效,使得消息分发能够持续,直到所有可用的磁盘被挂起(pending)的消息耗尽(无论是持久化的还是配置了非持久化的)。要这样做,你可以使用消息游标(Message Cursors)。

System usage系统占用

You can also slow down producers via some attributes on the<systemUsage>element. Take a look at the following example:

你还可以通过<systemUsage>元素的一些属性来减慢生产者。来看一眼下面的例子:

<systemUsage>
  <systemUsage>
    <memoryUsage>
      <memoryUsage limit="64 mb" />
    </memoryUsage>
    <storeUsage>
      <storeUsage limit="100 gb" />
    </storeUsage>
    <tempUsage>
      <tempUsage limit="10 gb" />
    </tempUsage>
  </systemUsage>
</systemUsage>

You can set limits of memory forNON_PERSISTENTmessages, disk storage forPERSITENTmessages and total usage for temporary messages, the broker will use before it slowdown producers.Using the default settings shown above, the broker will block thesend()call until some messages are consumed and space becomes available on the broker.The default values are shown above, you will probably need to increase these values for your environment.

你可以为非持久化的消息(NON_PERSISTENTmessages)设置内存限制,为持久化消息(PERSISTENTmessages)设置磁盘空间,以及为临时消息设置总的空间,代理将在减慢生产者之前使用这些空间。使用上述的默认设置,代理将会一直阻塞sen()方法的调用,直至一些消息被消费,并且代理有了可用空间。默认值如上例所述,你可能需要根据你的环境增加这些值。

分享到:
评论

相关推荐

    springboot整合activemq 生产者 一对一,一对多

    springboot整合 activeMq 生产者 发送消息 包含队列模式点对点发送消息 以及 主题模式一对多发送消息 这是生产者的demo producer; 需要配合消费者的demo consumer 使用

    ActiveMQ生产者

    **ActiveMQ生产者详解** ActiveMQ是Apache组织开发的一个开源消息中间件,它遵循Java Message Service(JMS)规范,提供了高效、可靠的异步通信能力。在分布式系统中,ActiveMQ作为消息代理,允许应用程序之间通过...

    基于SpringBoot的ActiveMQ生产者/消费者

    在本项目中,我们探讨的是如何使用SpringBoot集成Apache ActiveMQ来构建一个生产者和消费者的应用。SpringBoot以其简洁的配置和快速启动特性,成为现代Java应用开发的首选框架之一,而ActiveMQ则是流行的消息中间件...

    activemq生产者和消费者案例代码.zip

    3. **生产者(Producer)** 生产者是创建和发送消息到消息队列的角色。在Java中,这通常涉及创建一个`ConnectionFactory`,然后通过这个工厂创建一个`Connection`对象,接着创建一个`Session`,在会话中创建`...

    activeMQ生产者和消费者代码

    在分布式系统中,ActiveMQ作为消息代理,负责接收、存储和转发消息,从而实现生产者与消费者之间的解耦。 生产者和消费者是JMS中的核心概念。生产者是发送消息的应用,而消费者则是接收这些消息的应用。在ActiveMQ...

    spring 整合 activemq 生产者和消费者 案例源码

    Spring整合ActiveMQ是Java消息服务(JMS)在Spring框架中的应用,用于实现生产者与消费者的解耦。在这个案例中,我们将深入探讨如何配置和使用这两个组件,以便于理解它们的工作原理。 首先,ActiveMQ是Apache软件...

    自己实现的 ActiveMQ 多线程客户端 包含生产消息客户端和消费者消息客户端

    - **Amq_Producer.cpp**:这是单线程消息生产者的实现,可能包含创建连接、创建生产者对象、构建消息和发送消息的代码。 - **Amq_Producer_mt.cpp**:扩展了 Amq_Producer.cpp,增加了多线程支持,每个线程独立...

    springboot整合activemq 消费者 ACK手动确认 &消息重发

    springboot整合 activeMq 消费者 消费接收消息 包含队列模式点对点发 以及 主题模式一对多 这是消费者的demo consumer 。 里面有消息重发机制,手动确认ACK模式。...配合 producer 生产者demo使用。

    activemq实战项目,同ssh框架整合(生产者+消费者)

    - **生产者**:生产者负责创建和发送消息到ActiveMQ。在SSH项目中,生产者可能是一个服务或控制器,它将业务数据包装成消息并发送到队列或主题。 - **消费者**:消费者订阅队列或主题,接收并处理消息。消费者可能...

    activeMQ 详细教程与源码(包含消费者与生产者)

    **消息生产者**(Producer)是发送消息的组件,通常在业务处理完成后,将结果或者事件封装为消息发送到消息队列。在 Spring 中,可以通过 `JmsTemplate` 或者 `MessageProducer` 接口来实现。`springMQProducer.rar`...

    activeMQ Demo

    - **生产者(Producer)**: 负责创建并发送消息到消息队列。 - **消费者(Consumer)**: 从消息队列中接收并处理消息。 - **主题(Topic)**: 支持发布/订阅模式,一个主题可以有多个订阅者,当生产者发送一条...

    activeMQ简单入门案例

    - 实现生产者代码,创建一个连接工厂,连接到ActiveMQ服务器,并创建一个Producer发送消息: ```java import org.apache.activemq.ActiveMQConnectionFactory; // 创建连接工厂 ActiveMQConnectionFactory ...

    spring-boot-activemq-producer

    spring-boot-activemq-producer 源码

    activemqactivemq

    发送消息涉及到创建一个JMS生产者。首先,你需要创建一个Session,然后根据需求创建一个Destination(队列或主题)。接下来,创建一个MessageProducer,并使用它来发送消息: ```java MessageProducer producer = ...

    ActiveMQ集群及生产者和消费者Java代码.zip

    在这个“ActiveMQ集群及生产者和消费者Java代码”压缩包中,我们可以探讨以下几个关键知识点: 1. **ActiveMQ集群**:ActiveMQ的集群能力允许多个服务器形成一个逻辑单元,提供高可用性和负载均衡。当一个消息代理...

    activemq生产消费的Demo

    java整合activemq的demo,生产者和消费者两个方法。结合自带的工具http://192.168.1.106:8161。来查看消息传递情况

    ActiveMQ整合Spring(多消费者)

    在IT行业中,消息队列(Message Queue)是分布式系统中常用的一种技术,它能有效地解耦各个服务...通过合理配置和使用消息队列,可以提高系统的响应速度,降低服务间的耦合,同时为故障恢复和流量控制提供了有效手段。

    activemq限流排队

    标题中的“activemq限流排队”指的是Apache ActiveMQ中的一种流量控制策略,它用于管理消息生产者和消费者之间的消息传递速率,以防止系统过载。ActiveMQ是Apache软件基金会开发的一个开源的消息中间件,它遵循Java...

    自己写的ActiveMQ的Demo例子

    1. **生产者(Producer)**:生产者是创建和发送消息到消息队列的实体。在这个Demo中,生产者将负责创建消息并将其发送到特定的队列或主题。 2. **消费者(Consumer)**:消费者是从消息队列接收和处理消息的实体。...

    Activemq入门实例.pdf

    - 在Java中创建ActiveMQ生产者涉及到使用JMS API。 - 首先需要创建ConnectionFactory对象,这里使用ActiveMQ提供的ActiveMQConnectionFactory类。 - 然后通过ConnectionFactory创建连接(Connection)并启动。 -...

Global site tag (gtag.js) - Google Analytics