ActiveMQ(二十五)--ActiveMQ的集群1
Queue consumer cluster和Broker cluster
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class QueueSender1 {
public static void main(String[] args) throws JMSException, InterruptedException {
// ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
"failover:(tcp://192.168.25.128:61616,tcp://192.168.25.128:61716)?randomize=true");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("my-queue");
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 30; i++) {
TextMessage message = session.createTextMessage("message999--" + i);
producer.send(message);
}
session.commit();
session.close();
connection.close();
}
}
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class QR1 {
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 < 1; i++) {
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage txtMsg = (TextMessage) message;
try {
System.out.println("QR111===" + txtMsg.getText());
session.commit();
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
}
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class QR2 {
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61716");
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 < 1; i++) {
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage txtMsg = (TextMessage) message;
try {
System.out.println("QR222===" + txtMsg.getText());
session.commit();
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
}
先启动QR1和QR2。然后,再启动QueueSender1。
此时,c1和c2均衡的消费由brokerA和borkerB组成的集群里的消息。
修改
先启动QR1和QR2。然后,再启动QueueSender1。
查看结果,c1和c2依然是均衡的消费。
这是因为
修改conf里面的activmq.xml
重新启动mq
先启动QR1和QR2。然后,再启动QueueSender1。