Apache ServiceMix

속도가 느린 컨슈머는 일단 채워진 RAM에 오래된 메시지를 유지하기 위해 프로듀서에 속도를 늦추도록 브로커에 강제할 수 있고 빠른 컨슈머가 속도가 느려지게 되기 때문에 비영속성 토픽에서 문제가 발생 할 수 있다. 이후에 우리가 구현 할수 있는 하나의 옵션은 디스크에 스플링(디스크에 쓰기와 동시에 처리) 하는 것이다.  그러나 디스크에 스플링 하는 것 역시 속도가 빠른 컨슈머를 속도를 감소 시킨다.

현재 브로커의 프리패치 버퍼에 추가된 컨슈머를 위해 브로커가 유지하게 되는 일치하는 메시지의 최대 수를 환경 구성하는 전략을 가지고 있다. 최대 값에 도달 할 때, 새 메시지가 들어 오기 때문에 오래된 메시지들은 폐기 된다. 이것은 현재 메시지를 위해 RAM을  유지하고 속도가 느린 컨슈머에 메시지를 보내는 것을 유지하는 것을 허용한다. 그러나 오래된 메시지는 폐기한다.

지연 메시지 제한 전략(Pending Message Limit Strategy)

목적지 Map에 PendingMessageLimitStrategy 구현체 클래스를 환경 구성할 수 있다. 그래서 토픽 네임스페이스의 다른 영역이 속도가 늦은 컨슈머를 다루기 위한 다른 지연 전략을 가질 수 있다. 예를 들면,  매우 부피가 큰가격을 위해 지연 전략을 사용하기를 원할지 모른다. 그러나 매우 작은 부피를 가지는 주문과 거래를 위해 오래된 메시지는 폐기하는 것을 원하지 않을지 모른다.

이 전략은 컨슈머를 위해 RAM에 유지 되어야 하는 지연되는 메시지의 최대 량을 계산한다. (위 프리패치 사이즈). 0 값은 프리패치 량보다 더 이상의 메시지를 유지 하지 않는 다는 것을 의미한다. 0보다 큰 값은 메시지의 량에 들어 온 새 메시지로 오래된 메시지를 폐기하면서 프리패치 량보더 더 메시지를 유지 하게 될 것이다.  -1 값은 메시의 폐기를 비활성화 한다. 

현재 이 전략은 다른 두 구현체가 있다. :

• 정해진 상수 지연 메시지 제한 전략 (ConstantPendingMessageLimitStrategy)
• 프리패치 비율 지연 메시지 제한 전략 (PrefetchRatePendingMessageLimitStrategy)

정해진 상수 지연 메시지 제한 전략 (ConstantPendingMessageLimitStrategy)

이 전략은 모든 컨슈머를 위해 정해진 상수 제한값을 사용한다.  (위 프리패치 사이즈).

예제:

프리패치 비율 지연 메시지 제한 전략 (PrefetchRatePendingMessageLimitStrategy)

이 전략은 컨슈머들 프리패치 사이즈의 곱셈을 사용하여 지연 메시지의 최대 량을 계산한다. 그래서 예를 들면 각각의 컨슈머를 위한 프리패치 수를 2.5배 유지 할 수 있다. 

예제:

제한을 환경 구성하기 위해 프리패치 정책 사용하기 

JMS 클라이언트는 지속성 또는 비지속성 큐와 토픽을 위한 다양한 프리패치 정책을 환경 구성하기 위해 사용될 수 있는 프리패치 정책 을 가진다. 이 프리패치 정책은 또한 커넥션 단위 또는 컨슈머 단위로 maximumPendingMessageLimit 을 지정하는 것을 허용한다. 이 값을 환경 구성하는 하나의 작은 차이가 있다; OpenWire 같은 non-JMS 클라이언트로 동작을 단순화 하기 위해,  0값이 무시된다.; 그래서 환경구성 될수 있는 가장 낮은 값은 1이다.

퇴출 정책 환경 구성하기

속도가 느린 컨슈머에 메시지가 퇴출되는 것을 결정할때 사용하는 MessageEvictionStrategy  전략이 있다. 

하나는 가장 오래된 메시지를 퇴출 하는 것이 기본 값이고 설정은 다음과 같다:

그러나, 몇몇 어플리케이션이 퇴출을 위한 메시지를 선택하는 특별한 방법을 사용하기 위해 사용할 수 있다. 예를들면, 만약 가격을 업데이트 할 시장 데이타를 보낸다면, 가장 오래된 메시지가 아닌 오래된 가격 값을 찾는 것을 원할지 모른다.

예제:

propertyName 은 가격을 지정한 JMS 메시지 프로퍼티이다. 

역자주) propertyName 에 지정된 이름에 가장 오래된 메시지를 퇴출 한다.

다른 하나의 옵션은 우선 순위가 가장 낮은 메시지로 가장 오래된 메시지를 사용할 수있다. 그러므로 만약 몇몇 높은 우선순위 메시지가 있다면 심지어 새 메시지라고 하더라도 우선순위가 낮은 메시지가 퇴출된다. 

예제

다음 예지는 ActiveMQ 브로커의 환경 구성 파일을 보여준다. Notice that for topics in the PRICES.> wildcard range the pendingMessageLimitStrategyproperty is set to only keep around 10 messages for each consumer above their prefetch buffer size.

<beans xmlns="http://www.springframework.org/schema/beans"
            xmlns:amq="http://activemq.apache.org/schema/core"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
            http://activemq.apache.org/schema/core  http://activemq.apache.org/schema/core/activemq-core.xsd">

<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
    <broker xmlns="http://activemq.apache.org/schema/core" persistent="false" brokerName="${brokername}">
      <!-- lets define the dispatch policy -->
      <destinationPolicy>
        <policyMap>
          <policyEntries>

            <policyEntry topic="FOO.>">
              <dispatchPolicy>
                <roundRobinDispatchPolicy/>
              </dispatchPolicy>
              <subscriptionRecoveryPolicy>
                <lastImageSubscriptionRecoveryPolicy/>
              </subscriptionRecoveryPolicy>
            </policyEntry>

            <policyEntry topic="ORDERS.>">
              <dispatchPolicy>
                <strictOrderDispatchPolicy/>
              </dispatchPolicy>
              <!-- 1 minutes worth -->
              <subscriptionRecoveryPolicy>
                <timedSubscriptionRecoveryPolicy recoverDuration="60000"/>
             </subscriptionRecoveryPolicy>
          </policyEntry>
  
          <policyEntry topic="PRICES.>">
            <!-- lets force old messages to be discarded for slow consumers -->
            <pendingMessageLimitStrategy>
              <constantPendingMessageLimitStrategy limit="10"/>
            </pendingMessageLimitStrategy>
            <!-- 10 seconds worth -->
            <subscriptionRecoveryPolicy>
              <timedSubscriptionRecoveryPolicy recoverDuration="10000"/>
            </subscriptionRecoveryPolicy>
          </policyEntry>
  
          <policyEntry tempTopic="true" advisoryForConsumed="true"/>
          <policyEntry tempQueue="true" advisoryForConsumed="true"/>

        </policyEntries>
      </policyMap>
    </destinationPolicy>
  </broker>
</beans>

사용 팁

만약 특정한 컨슈머가 속도가 느리다면 빠른 컨슈머보다 프리패치 사이즈를 더 작게 설정하는 것이 좋다!

속도가 느린 컨슈머 상태 모니터링 하기

활성화된 구독의 상태를 보기 위해 JMX 콘솔을 사용 할 수 있다. 이 방법은 TopicSubscriptionViewMBean에 다음 통계를 보는것을 허용한다.