title: Extremely large ActiveMQ deployment performance issues
body: <p>We are planning to deploy an ActiveMQ farm capable of supporting 4+ million concurrent connections across many servers.  We are NOT using clustering based on our design.  The clients connect using MQTT and subscribe/consume to/from their own topic and publish to a shared virtual topic.  We are thinking of 100k connections and 20k msg/s per server.  I am nowhere near reaching that amount.  I have been successful connecting that amount, however its throughput that is hurting me.</p>
  
  <p>To test, I am using a multithreaded client based around the Async (callback not future) version of the FuseSource MQTT client.  Using that client I am able to put a load on the system great enough that I see queue backlog.  I am getting the queue backlog using “java -jar activemq-all-5.13.0.jar dstat”.</p>
  
  <p>I have tweaked the server in many ways but fail to be able to utilize all resources.  Iostat show low %util and top shows available cpu resources.</p>
  
  <p>My server is running on a AWS m3.2xlarge (8vCPU, 30G ram).</p>
  
  <p>I am starting activemq with “java –Xms12G –Xmx20G –jar …”</p>
  
  <p>Here is how I am starting my broker.  Note that I have altered many setting but they don't seem to matter.</p>
  
  <pre><code>@Bean
  public BrokerService getBrokerService() throws Exception {
      String hostname= InetAddress.getLocalHost().getHostName();
  
      // Create broker
      BrokerService broker = new BrokerService();
      broker.setBrokerName("ActiveMQ"+hostname);
      broker.setUseShutdownHook(true);
      broker.setAdvisorySupport( false );
      broker.setUseJmx(true);
      broker.setDeleteAllMessagesOnStartup( true );
      broker.setPopulateJMSXUserID( true );
      broker.setPopulateUserNameInMBeans( true );
      broker.setUseAuthenticatedPrincipalForJMSXUserID( true );
      broker.setUseVirtualTopics( true );
  
      // I have altered these but they don't seem to change anything.
      broker.setDedicatedTaskRunner( true );
      broker.setEnableStatistics( true );
      broker.setStartAsync( true );
  
      // Enable JMX bean
      final ManagementContext managementContext = new ManagementContext();
      managementContext.setCreateConnector(true);
      broker.setManagementContext( managementContext );        
  
      // Enable LevelDB
      broker.setPersistent( true );
      File dataFileDir = new File("/activemq/data/leveldb);
      LevelDBStore adapter = new LevelDBStore();
      adapter.setDirectory(dataFileDir);
      broker.setPersistenceAdapter(adapter);
  
      // Turn flow control off
      PolicyEntry policy = new PolicyEntry();
      policy.setOptimizedDispatch( true );
      policy.setProducerFlowControl( false );
      policy.setGcInactiveDestinations( true );
      DispatchPolicy dp = new RoundRobinDispatchPolicy();
      policy.setDispatchPolicy( dp );
      PolicyMap policyMap = new PolicyMap();
      policyMap.setDefaultEntry( policy );
      broker.setDestinationPolicy( policyMap );
  
       // Set system usage
      SystemUsage memoryManager = new SystemUsage();
      MemoryUsage memoryUsage = new MemoryUsage();
      memoryUsage.setPercentOfJvmHeap( 75 );
      memoryManager.setMemoryUsage( memoryUsage );
  
      StoreUsage storeUsage = new StoreUsage();
      storeUsage.setLimit( 100*ONE_GB );
      memoryManager.setStoreUsage( storeUsage );
  
      TempUsage tempDiskUsage = new TempUsage();
      tempDiskUsage.setLimit( 100*ONE_GB );
      memoryManager.setTempUsage( tempDiskUsage );
      broker.setSystemUsage( memoryManager );
  
      // Add transports
      TransportConnector openwireconnector = new TransportConnector();
      openwireconnector.setName( "openwire" );
      openwireconnector.setUri(new URI("tcp://0.0.0.0:61616"));
      broker.addConnector(openwireconnector);
  
      TransportConnector mqttnioconnector = new TransportConnector();
      mqttnioconnector.setName( "mqtt+nio" );
      mqttnioconnector.setUri(new URI("mqtt+nio://0.0.0.0:1883"));
      broker.addConnector(mqttnioconnector);
  
      TransportConnector wsconnector = new TransportConnector();
      wsconnector.setName( "ws" );
      wsconnector.setUri(new URI("ws://0.0.0.0:61614"));
      broker.addConnector(wsconnector);
  
      // Start
      broker.start();
      broker.waitUntilStarted();
     return broker;
     }
  }
  </code></pre>
  
  <p>I am using a spring JMS listener in another server which connects to ActiveMQ.</p>
  
  <pre><code>@Component
  public class MqttMessageListener {
      @JmsListener(destination = "Consumer.A.VirtualTopic.Message")
      public void processMessage(BytesMessage bytesMessage) {
       // I consume and discard here just for testing.  No work is being done to isolate issue.    
      }
  }
  
  @Configuration
  @EnableJms
  public class AppConfig {
      @Bean
      public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
         DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
      factory.setConnectionFactory(connectionFactory);
      factory.setConcurrency(“3-10”);  // I have tried many variation including 300-1000 although it seems to top out at 500.
      factory.setAutoStartup( true );
      return factory;
      }
  }
  </code></pre>
  
  <p>With the above MqttMessageListener.processMessage method I am only able to consume at around 15k msgs/sec.  With 30+% cpu idle and low disk/network utilization I am unsure why I can't do better.</p>
  
  <p><strong>QUESTIONS:</strong></p>
  
  <ol>
  <li>How can I find the limiting resource?</li>
  <li>Is their an easy way to determine if this is an issue with ActiveMQ delivery or the consumption side?  Both are Spring Boot applications.</li>
  </ol>