2

我想创建一个消费者来堆叠到达的消息，然后等待：

  • 直到n消息到达。
  • t秒过去了。

处理整个消息堆栈。

预取不是我想要的。我真正需要的是一起处理消息。

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")
4

1 回答 1

1

这是一个例子

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();
    }
}
于 2017-04-24T11:58:51.057 回答