Re: No performance improvement using PooledConnectionFactory (AMQ 5.12.1).

[Frizz]

On Tue, Feb 23, 2016 at 12:25 PM, Robbie Gemmell <robbie.gemm...@gmail.com>
wrote:

> I can't say for sure from your summarised description and partial
> code, but it seems like you may only ever create 1 connection object
> with the factory. If so, explicitly creating multiple connections and
> splitting your producers across them would likely improve performance.
>


I am calling "Connection connection =
pooledConnectionFactory.createConnection(); connection.start();" for every
Thread / Producer - so 250 times.

At least for my pooled connection is should result in 8 "real" connections,
no?


Here is my test class:

public class ProducerMT {

    private static final long NUM_MESSAGES_PER_PRODUCER = 1000;
    private final static int NUM_PRODUCERS = 250;

    private static String connectionUrl = "tcp://localhost:61616";
    private MessageProducer producer = null;
    private Session session = null;

    private static ConnectionFactory factory = null;

    public static void setPooledConnectionFactory() {
        PooledConnectionFactory pcf = new
PooledConnectionFactory(connectionUrl);
        pcf.setMaxConnections(8);
        pcf.setMaximumActiveSessionPerConnection(500);
        factory = pcf;
    }


    public void sendMessage(String topicName, Event event) {

        if(producer == null) {

            try {
                Connection connection = factory.createConnection();
                connection.start();
                session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
                producer = session.createProducer(null);

            } catch (JMSException exp) {
                System.err.println("Error creating
connection/session/producer: " + exp);
                exp.printStackTrace();
            }
        } else

        try {
            final Topic topic = session.createTopic(topicName);
            final ObjectMessage message =
session.createObjectMessage(event);

            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            producer.send(topic, message);
        } catch (JMSException exp) {
            System.err.println("Error sending message: " + exp);
            exp.printStackTrace();
        }

    }


    private CompletableFuture<Integer> sendMessages(int eventId) {
        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(
            () -> {
                int i;
                for (i = 0; i < NUM_MESSAGES_PER_PRODUCER; i++) {
                    sendMessage("topic_" + eventId, new Event());
                }
            return i;
        });
        return cf;
    }

    public static void main(String[] args) {

        setPooledConnectionFactory();

        List<CompletableFuture<Integer>> returnValues = new ArrayList<>();

        ProducerMT producers[] = new ProducerMT[NUM_PRODUCERS];
        for (int i = 0; i < NUM_PRODUCERS; i++) {
            producers[i] = new ProducerMT();
        }

        long startTime = System.nanoTime();
        try {
            for (int i = 0; i < NUM_PRODUCERS; i++) {
                returnValues.add(producers[i].sendMessages(i));
            }

            for (int i = 0; i < NUM_PRODUCERS; i++) {
                returnValues.get(i).get();
            }

        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        long elapsedTime = System.nanoTime() - startTime;
        double seconds = (double)elapsedTime / 1000000000.0;

    }
}