How to stack received messages and process them in batch

Question

I'd like to create a consumer which stacks arriving messages, and waits:

until n messages have arrived.
t seconds are elapsed.

to process the whole stack of messages.

Pre-fetching is not what I'm looking for. What I really need is to process messages together.

class MyListener(stomp.ConnectionListener):

    def on_message(self, headers, body):

        print ("Just received ONE message\n"
               "I should wait for n-1 others\n"
               "or t seconds before processing")

I think what you need is a queue implementation in your consumer, I am not sure how it is done in python but try searching for queue implementation and use it as per your requirement. — hagrawal7777, Apr 25 '17 at 17:43

score 1 · Answer 1 · answered Apr 24 '17 at 11:58

here an example

import java.util.LinkedList;
import java.util.List;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageConsumer;

public class SimpleConsumerClientAcknowledge {

    public static void main(String[] args) throws JMSException {
        List<TextMessage> messages = new LinkedList<>();
        Connection conn = null;
        try {
            ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
                    "tcp://localhost:61617?jms.prefetchPolicy.all=200");
            conn = cf.createConnection("admin", "admin");
            Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer) session
                    .createConsumer(session.createQueue("Q"));
            conn.start();
            TextMessage msg = null;
            // MAX_MESSAGES have to be < prefetchSize / 2 -->
            // jms.prefetchPolicy.all=200
            // Once the broker has dispatched a prefetch limit number of
            // messages to a consumer it will not dispatch any more messages to
            // that consumer until the consumer has acknowledged at least 50% of
            // the prefetched messages
            int MAX_MESSAGES = 100;
            long MAX_WAIT = 60000;
            long millis = System.currentTimeMillis();
            while ((msg = (TextMessage) consumer.receive(5000)) != null) {
                if (msg != null) {
                    messages.add(msg);
                }
                if (messages.size() == MAX_MESSAGES || (System.currentTimeMillis() - millis >= MAX_WAIT)) {
                    millis = System.currentTimeMillis();
                    treatMessages(messages);
                    // because session is created with
                    // Session.CLIENT_ACKNOWLEDGE as an acknowledgeMode consumer
                    // need to acknowledge manually received messages
                    consumer.acknowledge();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (Exception e) {
                }
            }
        }
    }

    private static void treatMessages(List<TextMessage> messages) {
        // TODO Auto-generated method stub
        messages.clear();
    }
}

How to stack received messages and process them in batch

1 Answers1