主从共享部署分为Shared FileSystem Master-Slave 和 Shared DatabaseMaster-Slave，数据源分别为文件系统（如 NAS）和数据库，这里以 MySQL 数据库为例。
1. 添加数据库驱动
为了让 ActiveMQ 支持数据库，先要在${ACTICEMQ_HOME}/lib/extra添加数据库的驱动包
[root@localhost extra]# pwd
/var/activemq/lib/extra
[root@localhost extra]# ls
mqtt-client-1.15.jar mysql-connector-java-5.1.47.jar
2. 修改配置文件
开打 ${ACTICEMQ_HOME}/conf/activemq.xml
在 broker 节点 添加 persistent="true" 使其支持持久化
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" persistent="true" dataDirectory="${activemq.data}" schedulerSupport="true">
修改持久化适配器为数据库
<persistenceAdapter>
<jdbcPersistenceAdapter dataSource="#mysql-ds" useDatabaseLock="false" transactionIsolation="4"/>
</persistenceAdapter>
在 broker 节点后添加数据源
<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://192.168.0.100:3306/test_activemq?relaxAutoCommit=true"/>
<property name="username" value="root"/>
<property name="password" value="root123"/>
<property name="poolPreparedStatements" value="true"/>
</bean>
3. 启动多个 ActiveMQ
启动多个配置文件相同的 ActiveMQ，多个 ActiveMQ 开启时，同一时间只有一个服务器可以拿到 MySQL 的连接。
4. 代码
消费者
public class ConsumerFailover {
public static void main(String[] args) throws InterruptedException {
// 非failover的公共参数配置通过nested.*，例如 failover:(...)?nested.wireFormat.maxInactivityDuration=1000
// ?randomize=false 随机选择，默认是顺序
// 指定优先切换 failover:(tcp://host1:61616,tcp://host2:61616,tcp://host3:61616)?priorityBackup=true&priorityURIs=tcp://local1:61616,tcp://local2:61616
// maxReconnectDelay重连的最大间隔时间(毫秒)
String brokerUrl = "failover:(tcp://192.168.0.102:61616,tcp://192.168.0.106:61616)?initialReconnectDelay=100";
ConsumerThread queue1 = new ConsumerThread(brokerUrl, "queue1");
queue1.start();
queue1.join();
}
}
class ConsumerThread extends Thread {
String brokerUrl;
String destinationUrl;
public ConsumerThread(String brokerUrl, String destinationUrl) {
this.brokerUrl = brokerUrl;
this.destinationUrl = destinationUrl;
}
@Override
public void run() {
ActiveMQConnectionFactory connectionFactory;
Connection conn;
Session session;
MessageConsumer consumer;
try {
// brokerURL http://activemq.apache.org/connection-configuration-uri.html
// 1、创建连接工厂
connectionFactory = new ActiveMQConnectionFactory(this.brokerUrl);
// 2、创建连接对象
conn = connectionFactory.createConnection();
conn.start(); // 一定要启动
// 3、创建会话（可以创建一个或者多个session）
session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4、创建点对点接收的目标，queue - 点对点
Destination destination = session.createQueue(destinationUrl);
// 5、创建消费者消息 http://activemq.apache.org/destination-options.html
consumer = session.createConsumer(destination);
// 6、接收消息
consumer.setMessageListener(message -> {
try {
if (message instanceof TextMessage) {
System.out.println("收到文本消息：" + ((TextMessage) message).getText());
} else {
System.out.println(message);
}
} catch (JMSException e) {
e.printStackTrace();
}
});
} catch (JMSException e) {
e.printStackTrace();
}
}
}
启动后客户端会按顺序去连接服务器，连接上了第一个
INFO | Successfully connected to tcp://192.168.0.106:61616
生产者
public class Producer {
public static void main(String[] args) {
// 添加两个中间件，可以实现故障转移
String brokerUrl = "failover:(tcp://192.168.0.102:61616,tcp://192.168.0.106:61616)?initialReconnectDelay=100";
new ProducerThread(brokerUrl, "queue1").start();
}
static class ProducerThread extends Thread {
String brokerUrl;
String destinationUrl;
public ProducerThread(String brokerUrl, String destinationUrl) {
this.brokerUrl = brokerUrl;
this.destinationUrl = destinationUrl;
}
@Override
public void run() {
ActiveMQConnectionFactory connectionFactory;
Connection conn;
Session session;
try {
// 1、创建连接工厂
connectionFactory = new ActiveMQConnectionFactory(brokerUrl);
// 2、创建连接对象md
conn = connectionFactory.createConnection();
conn.start();
// 3、创建会话
session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4、创建点对点发送的目标
Destination destination = session.createQueue(destinationUrl);
// 5、创建生产者消息
MessageProducer producer = session.createProducer(destination);
// 设置生产者的模式，有两种可选 持久化 / 不持久化
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
// 6、创建一条文本消息
String text = "Hello world!";
TextMessage message = session.createTextMessage(text);
for (int i = 0; i < 1; i++) {
// 7、发送消息
producer.send(message);
}
// 8、 关闭连接
session.close();
conn.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
运行以下生产者的代码，可以看到消费者接收到了
INFO | Successfully connected to tcp://192.168.0.106:61616
收到文本消息：Hello world!
5. 模拟故障
通过手动关闭 ActiveMQ 模拟故障场景，消费者断开后连接到第二个服务器
java.io.EOFException
at java.io.DataInputStream.readInt(DataInputStream.java:392)
at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:268)
at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:240)
at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:232)
at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215)
at java.lang.Thread.run(Thread.java:748)
INFO | Successfully connected to tcp://192.168.0.106:61616
网友评论