













配置过程
1. 节点准备
mkdir activemq创建目录
cp -rf apache-activemq-5.15.0 activemq/activemq-a
cp -rf apache-activemq-5.15.0 activemq/activemq-b
cp -rf apache-activemq-5.15.0 activemq/activemq-c
cd activemq
mkdir kahadb
2. 配置a节点
cd activemq-a/
cd conf/
vim activemq.xml
vim jetty.xml:配置管理端口号,a节点使用默认端口,无须配置
3. 配置b节点
vim activemq.xml
配置网络连接器
配置持久化存储路径
配置服务端口
vim jetty.xml
配置管理端口号
4. 配置c节点
vim activemq.xml
配置网络连接器
配置持久化存储路径
配置服务端口
vim jetty.xml
配置管理端口号
5.启动服务
回到activemq目录,分别启动a,b,c三个节点
./activemq-a/bin/activemq start
./activemq-b/bin/activemq start
./activemq-c /bin/activemq start
检查是否都启动成功
ps -ef |grep activemq
检查是否对外提供服务,即端口是否被监听(占用)
netstat -anp |grep 61616
netstat -anp |grep 61617
netstat -anp |grep 61618
检查发现61618即c节点没有提供服务,但是c节点的进程是启动成功了的。因为b节点和c点击是master/slave配置,现在b节点获取到了共享文件夹的所有权,所以c节点正在等待获得资源,并且提供服务。即c节点在未获得资源之前,是不提供服务的。
测试,把b节点杀掉,看c节点能不能提供61618的服务
./activemq-b/bin/activemq stop
netstat -anp |grep 61618
./activemq-b/bin/activemq start
netstat -anp |grep 61617
检查发现,重新启动b节点后,b节点61617端口并没有提供服务,是因为现在b节点成为了slave节点,而c节点成为了master节点。所以,现在b节点启动了,但是它并不对外提供服务。只有当c节点出现问题后,b节点才对外提供服务。
6.通过代码测试集群配置是否生效
生产者
package com.myimooc.jms.queue;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class AppProducerTest {
private static final String URL = "failover:(tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)?randomize=true";
private static final String QUEUE_NAME = "queue-test";
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(QUEUE_NAME);
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < 100; i++) {
TextMessage textMessage = session.createTextMessage("test" + i);
producer.send(textMessage);
System.out.println("消息发送:" + textMessage.getText());
}
connection.close();
}
}
消费者
package com.myimooc.jms.queue;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class AppConsumerTest {
private static final String URL = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)?randomize=true";
private static final String QUEUE_NAME = "queue-test";
public static void main(String[] args) throws JMSException {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(QUEUE_NAME);
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage)message;
try {
System.out.println("接收消息:" + textMessage.getText());
} catch (JMSException e) {
System.out.println("接收消息异常:");
e.printStackTrace();
}
}
});
}
}
运行生产者,然后到管理界面查看消息发送到了那里
http://127.0.0.1:8161
http://127.0.0.1:8162
http://127.0.0.1:8163
查看发现,8162无法访问,是因为b节点是slave节点,不提供服务,消息都发送到了c节点
把8163即c节点宕掉后,运行消费者,查看消息是否能够使用
./activemq-c/bin/activemq stop