限制在ActiveMq主题中发送消息的订户的最大数量

Question

假设有一个ActiveMq实例，并且有100个客户端正在监听主题。 当在主题中发布新消息时，是否可以限制接收它的订阅者数量（仅示例10）？

如果现在有其他消息传递组件能够做到这一点，或者存在解决方案/合并最佳实践？

Answer 1

开箱即不可配置，但您可以创建一个插件来做这样的事情非常简单。

一个插件，拦截addConsumer，并且如果已经有太多订阅者会抛出一些SecurityException。它也可以从activemq.xml config进行配置。

请注意，这是一个快速和肮脏的例子，你可能需要加强它。

MaxSubscribersPlugin.java

import org.apache.activemq.broker.BrokerPluginSupport; 
import org.apache.activemq.broker.ConnectionContext; 
import org.apache.activemq.broker.region.Destination; 
import org.apache.activemq.broker.region.Subscription; 
import org.apache.activemq.command.ActiveMQDestination; 
import org.apache.activemq.command.ConsumerInfo; 

import java.util.Map; 

public class MaxSubscribersPlugin extends BrokerPluginSupport{ 

    private int maxSubscribers = 1000; 

    @Override 
    public Subscription addConsumer(ConnectionContext ctx, ConsumerInfo info) throws Exception { 
     ActiveMQDestination dest = info.getDestination(); 
     if (dest.isTopic() && !dest.getQualifiedName().contains("Advisory")) { // TODO better way to filter out Advisories 
      Map<ActiveMQDestination, Destination> destinations = ctx.getBroker().getBrokerService() 
                    .getAdminView().getBroker().getDestinationMap(); 
      Destination activeTopic = destinations.get(info.getDestination()); 
      if(activeTopic != null && activeTopic.getConsumers().size() >= getMaxSubscribers()) { 

       throw new SecurityException("Too many active subscribers on topic " 
         + info.getDestination().getPhysicalName() 
         + ". Max is " + getMaxSubscribers()); 
      } 
     } 

     return super.addConsumer(ctx, info); 
    } 

    public int getMaxSubscribers() { 
     return maxSubscribers; 
    } 

    public void setMaxSubscribers(int maxSubscribers) { 
     this.maxSubscribers = maxSubscribers; 
    } 

} 

装箱打包在一个.jar并把它在ActiveMQ中的lib文件夹。然后你可以配置它。

<plugins> 
      <bean xmlns="http://www.springframework.org/schema/beans" id="throttler" 
       class="my.package.MaxSubscribersPlugin"> 
      <property name="maxSubscribers" value="10"/> 
     </bean> 
    </plugins> 

然后，如果有太多的消费者建立联系 - 他们会得到一个异常：javax.jms.JMSSecurityException: Too many active subscribers on topic topicX. Max is 10

会有在ActiveMQ的日志条目记录以及：

WARN | Security Error occurred on connection to: tcp://127.0.0.1:52335, Too many active subscribers on topic topicX. Max is 10