ActiveMQ是一个开源的消息中间件,它支持多个消费者同时消费同一个队列。下面是一个简单的示例代码,演示了如何使用ActiveMQ实现一个队列多个消费者的场景。
首先,你需要引入ActiveMQ的相关依赖,例如在Maven项目中可以添加以下依赖:
org.apache.activemq
activemq-all
5.16.1
然后,你可以创建一个生产者,向队列发送消息。下面的代码展示了如何创建一个ActiveMQ的连接,创建一个Session,并通过Session创建一个Producer,然后发送消息到队列中。
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Producer {
public static void main(String[] args) throws JMSException {
// 创建ActiveMQ连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
// 创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
// 创建Session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination destination = session.createQueue("myQueue");
// 创建Producer
MessageProducer producer = session.createProducer(destination);
// 发送消息
for (int i = 0; i < 10; i++) {
TextMessage message = session.createTextMessage("Message " + i);
producer.send(message);
System.out.println("Sent message: " + message.getText());
}
// 关闭连接
session.close();
connection.close();
}
}
接下来,你可以创建多个消费者来消费队列中的消息。下面的代码展示了如何创建两个消费者,从队列中接收消息。
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Consumer {
public static void main(String[] args) throws JMSException {
// 创建ActiveMQ连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
// 创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
// 创建Session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination destination = session.createQueue("myQueue");
// 创建Consumer1
MessageConsumer consumer1 = session.createConsumer(destination);
consumer1.setMessageListener(message -> {
try {
System.out.println("Consumer1 received message: " + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
});
// 创建Consumer2
MessageConsumer consumer2 = session.createConsumer(destination);
consumer2.setMessageListener(message -> {
try {
System.out.println("Consumer2 received message: " + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
});
// 等待消息
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 关闭连接
session.close();
connection.close();
}
}
上述代码创建了两个消费者,它们分别监听同一个队列,并通过setMessageListener
方法设置消息监听器来接收消息。当有消息到达队列时,两个消费者会并发地接收并处理消息。
通过上述示例代码,你可以实现ActiveMQ的一个队列多个消费者的场景。你可以根据实际需求,创建更多的消费者来并发处理队列中的消息。