ActiveMQ(三十三)--Consumer高级特性1
一、Exclusive Consumer
二、Consumer Dispatche Async
三、Consumer Priority
四、Manage Durable Subscribers
五、Message Groups
测试:
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageProducer;
import javax.jms.*;
public class QueueGroupSender {
public static void main(String[] args) throws JMSException, InterruptedException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("my-queue");
ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(destination);
for (int i = 0; i < 3; i++) {
TextMessage message = session.createTextMessage("message---" + i);
message.setStringProperty("JMSXGroupID", "GroupA");
producer.send(message);
}
session.commit();
session.close();
connection.close();
}
}
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class QueueGroupReceiver {
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
"tcp://192.168.25.128:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("my-queue");
for (int i = 0; i < 2; i++) {
final MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage txtMsg = (TextMessage) message;
try {
System.out.println(consumer + "收到消息:" + txtMsg.getText());
session.commit();
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
}
先运行消费者,后运行生产者。
虽然,有两个消费者线程。但是,3条消息都让一个消费者线程消费掉了。
增加一个group
继续运行生产者。
两个消费者线程,别接收各自组的消息。