CODEWITHSUNDEEP
javajbossactivemq-classic
Nathan
Nathan

Reputation: 1

Increase number of dequeued messages processed at a time (ActiveMQ)

I've got an MDB running in JBoss AS7 consuming messages from a remote ActiveMQ 5.6 broker using the ActiveMQ JCA resource adapter.

Producer Code:

public void sendMessage(IDeliveryMessage message) throws Exception {

    Connection conn = null;
    Session session = null;
    MessageProducer producer = null;

    try {
        // Setup the connection & session
        ConnectionFactory cf = NamedObjectLocator.getConnectionFactory();
        conn = cf.createConnection();
        Destination destination = NamedObjectLocator.getDestination();
        session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        conn.start();
        // Send a text message
        producer = session.createProducer(destination);
        ObjectMessage om = session.createObjectMessage(message);

        String id = System.currentTimeMillis()+"";
        om.setJMSCorrelationID(id);
        System.out.println("Producer: " + id);

        producer.send(om);

    } catch (Exception e) {
        throw new Exception(e.getMessage(), e);
    } finally {
        if (producer != null) {
            producer.close();
        }
        if (session != null) {
            session.close();
        }
        if (conn != null) {
            conn.stop();
            conn.close();
        }
    }
}

Consumer Code:

import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.Message;
import javax.jms.MessageListener;

@MessageDriven(name = "OutBoundMDB", activationConfig = {
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/OutboundIntegrationQueue")})
public class OutBoundMDB implements MessageListener {

    public static final long serialVersionUID = 8609676375853959140L;

    public void onMessage(Message message) {
        try {

            System.out.println("Consumer: " + message.getJMSCorrelationID());

            Thread.sleep(20000L);

        } catch (Exception e) {
            LogManager.logError(e);
        }

    }

}

JBoss Standalone.xml: 

<subsystem xmlns="urn:jboss:domain:resource-adapters:1.0">
    <resource-adapters>
        <resource-adapter>
            <archive>
                activemq-rar.rar
            </archive>
            <transaction-support>XATransaction</transaction-support>
            <config-property name="ServerUrl">
                tcp://localhost:61616
            </config-property>
            <connection-definitions>
                <connection-definition class-name="org.apache.activemq.ra.ActiveMQManagedConnectionFactory" jndi-name="java:/JmsXA" enabled="true" use-java-context="true" pool-name="ActiveMQConnectionFactoryPool" use-ccm="true">
                    <xa-pool>
                        <min-pool-size>1</min-pool-size>
                        <max-pool-size>20</max-pool-size>
                    </xa-pool>
                </connection-definition>
            </connection-definitions>
            <admin-objects>
                <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:/queue/OutboundIntegrationQueue" enabled="true" use-java-context="true" pool-name="ActiveMQQueue.OutboundIntegrationQueue">
                    <config-property name="PhysicalName">queue/OutboundIntegrationQueue</config-property>
                </admin-object>
            </admin-objects>
        </resource-adapter>
    </resource-adapters>
</subsystem>

Now the log printed shows that Queue is only able to Dequeue approximately 8 to 10 messages at a time. Since I made the thread sleep for 20 seconds in consumer code, it was only able to process approx. 10 messages and then a gap of 20 seconds and then next 10 messages. Below is the log that shows the results:

2018-03-29 14:29:37,214 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348177214
2018-03-29 14:29:37,225 INFO  [stdout] (default-threads - 23) Consumer: 1522348177214
2018-03-29 14:29:37,433 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348177433
2018-03-29 14:29:37,437 INFO  [stdout] (default-threads - 24) Consumer: 1522348177433
2018-03-29 14:29:37,744 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348177744
2018-03-29 14:29:37,762 INFO  [stdout] (default-threads - 25) Consumer: 1522348177744
2018-03-29 14:29:37,956 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348177956
2018-03-29 14:29:37,958 INFO  [stdout] (default-threads - 26) Consumer: 1522348177956
2018-03-29 14:29:38,207 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348178207
2018-03-29 14:29:38,213 INFO  [stdout] (default-threads - 27) Consumer: 1522348178207
2018-03-29 14:29:38,413 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348178413
2018-03-29 14:29:38,421 INFO  [stdout] (default-threads - 28) Consumer: 1522348178413
2018-03-29 14:29:38,647 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348178647
2018-03-29 14:29:38,656 INFO  [stdout] (default-threads - 29) Consumer: 1522348178647


2018-03-29 14:29:38,856 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348178856
2018-03-29 14:29:39,060 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348179060
2018-03-29 14:29:39,265 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348179265
2018-03-29 14:29:39,583 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348179583
2018-03-29 14:29:39,872 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348179872
2018-03-29 14:29:40,143 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348180142
2018-03-29 14:29:40,451 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348180451
2018-03-29 14:29:40,716 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348180716
2018-03-29 14:29:41,041 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348181041
2018-03-29 14:29:41,254 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348181254
2018-03-29 14:29:41,447 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348181447
2018-03-29 14:29:41,682 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348181682
2018-03-29 14:29:41,911 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348181911
2018-03-29 14:29:42,098 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348182098
2018-03-29 14:29:42,300 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348182300
2018-03-29 14:29:42,497 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348182497
2018-03-29 14:29:42,834 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348182833
2018-03-29 14:29:43,043 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348183043
2018-03-29 14:29:43,257 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348183257
2018-03-29 14:29:43,459 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348183459
2018-03-29 14:29:43,669 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348183669
2018-03-29 14:29:43,936 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348183936
2018-03-29 14:29:44,133 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348184132
2018-03-29 14:29:44,329 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348184329
2018-03-29 14:29:44,551 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348184551
2018-03-29 14:29:44,789 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348184789
2018-03-29 14:29:45,012 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348185012
2018-03-29 14:29:45,274 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348185274
2018-03-29 14:29:45,473 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348185473
2018-03-29 14:29:45,721 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348185721
2018-03-29 14:29:46,104 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348186104
2018-03-29 14:29:46,388 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348186388
2018-03-29 14:29:46,586 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348186586
2018-03-29 14:29:46,789 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348186789
2018-03-29 14:29:46,985 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348186985
2018-03-29 14:29:47,178 INFO  [stdout] (http-/0.0.0.0:8080-3) Producer: 1522348187178

20 SECONDS DELAY


2018-03-29 14:29:57,771 INFO  [stdout] (default-threads - 32) Consumer: 1522348179265
2018-03-29 14:29:57,772 INFO  [stdout] (default-threads - 30) Consumer: 1522348178856
2018-03-29 14:29:57,775 INFO  [stdout] (default-threads - 33) Consumer: 1522348179583
2018-03-29 14:29:57,782 INFO  [stdout] (default-threads - 31) Consumer: 1522348179060
2018-03-29 14:29:57,784 INFO  [stdout] (default-threads - 34) Consumer: 1522348179872
2018-03-29 14:29:58,226 INFO  [stdout] (default-threads - 35) Consumer: 1522348180142


AGAIN 20 SECONDS DELAY


2018-03-29 14:30:17,790 INFO  [stdout] (default-threads - 36) Consumer: 1522348180451
2018-03-29 14:30:17,806 INFO  [stdout] (default-threads - 37) Consumer: 1522348180716
2018-03-29 14:30:17,807 INFO  [stdout] (default-threads - 39) Consumer: 1522348181254
2018-03-29 14:30:17,812 INFO  [stdout] (default-threads - 40) Consumer: 1522348181447
2018-03-29 14:30:17,815 INFO  [stdout] (default-threads - 44) Consumer: 1522348182300
2018-03-29 14:30:17,830 INFO  [stdout] (default-threads - 42) Consumer: 1522348181911
2018-03-29 14:30:17,833 INFO  [stdout] (default-threads - 43) Consumer: 1522348182098
2018-03-29 14:30:17,842 INFO  [stdout] (default-threads - 38) Consumer: 1522348181041
2018-03-29 14:30:17,848 INFO  [stdout] (default-threads - 41) Consumer: 1522348181682
2018-03-29 14:30:18,233 INFO  [stdout] (default-threads - 45) Consumer: 1522348182497




2018-03-29 14:30:37,793 INFO  [stdout] (default-threads - 36) Consumer: 1522348182833
2018-03-29 14:30:37,810 INFO  [stdout] (default-threads - 37) Consumer: 1522348183043
2018-03-29 14:30:37,815 INFO  [stdout] (default-threads - 39) Consumer: 1522348183459
2018-03-29 14:30:37,818 INFO  [stdout] (default-threads - 40) Consumer: 1522348183669
2018-03-29 14:30:37,855 INFO  [stdout] (default-threads - 38) Consumer: 1522348183257
2018-03-29 14:30:37,859 INFO  [stdout] (default-threads - 41) Consumer: 1522348183936
2018-03-29 14:30:37,867 INFO  [stdout] (default-threads - 47) Consumer: 1522348184329
2018-03-29 14:30:37,869 INFO  [stdout] (default-threads - 46) Consumer: 1522348184132
2018-03-29 14:30:37,871 INFO  [stdout] (default-threads - 48) Consumer: 1522348184551






2018-03-29 14:30:57,804 INFO  [stdout] (default-threads - 50) Consumer: 1522348185012
2018-03-29 14:30:57,804 INFO  [stdout] (default-threads - 49) Consumer: 1522348184789
2018-03-29 14:30:57,816 INFO  [stdout] (default-threads - 37) Consumer: 1522348185274
2018-03-29 14:30:57,821 INFO  [stdout] (default-threads - 39) Consumer: 1522348185721
2018-03-29 14:30:57,827 INFO  [stdout] (default-threads - 40) Consumer: 1522348186104
2018-03-29 14:30:57,861 INFO  [stdout] (default-threads - 38) Consumer: 1522348185473
2018-03-29 14:30:57,864 INFO  [stdout] (default-threads - 41) Consumer: 1522348186388
2018-03-29 14:30:57,873 INFO  [stdout] (default-threads - 47) Consumer: 1522348186789
2018-03-29 14:30:57,876 INFO  [stdout] (default-threads - 46) Consumer: 1522348186586
2018-03-29 14:30:57,881 INFO  [stdout] (default-threads - 48) Consumer: 1522348186985

I want the number of Dequeued messages at a time (which is approx. 10 right now) to be increased to 50 or may be more.

Upvotes: 0

Views: 886

Answers (1)

Justin Bertram
Justin Bertram

Reputation: 35217

Take a look at the ActiveMQ JCA RA activation specification documentation. You almost certainly need to tune maxSessions and maxMessagesPerSession to fit your performance goals.

You may also need to adjust the number of MDB instances in the pool maintained by the container (i.e. JBoss EAP in this case). Search your server config for mdb-strict-max-pool. By default it derives the pool size from the CPU count of the server, but you can use max-pool-size to set it explicitly. I believe the derivation uses the the formula: Runtime.getRuntime().availableProcessors() * 4 (see here).

Upvotes: 1

Related Questions