SharedFile System Master Slave(共享文件系统)做ActiveMQ集群

WINDOWS环境下：http://www.apache.org/dyn/closer.cgi?path=/activemq/apache-activemq/5.9.0/apache-activemq-5.9.0-bin.zip下载apache-activemq-5.9.0-bin.tar，解压两次，分别命名为activemq1、activemq2，修改activemq1/conf/activemq.xml中<persistenceAdapter>标签下的<kahaDB>文件路径<directory>，修改为E:/activemq_data,详情如下：

<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->

<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:${activemq.conf}/credentials.properties</value>
</property>
</bean>

<bean id="logQuery" class="org.fusesource.insight.log.log4j.Log4jLogQuery"
lazy-init="false" scope="singleton"
init-method="start" destroy-method="stop">
</bean>

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" useJmx="true">

<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry topic=">" producerFlowControl="false" memoryLimit="10mb">
<!-- The constantPendingMessageLimitStrategy is used to prevent
slow topic consumers to block producers and affect other consumers
by limiting the number of messages that are retained
For more information, see:

http://activemq.apache.org/slow-consumer-handling.html

-->

<dispatchPolicy>

<strictOrderDispatchPolicy/>
</dispatchPolicy>

<subscriptionRecoveryPolicy>

<lastImageSubscriptionRecoveryPolicy/>
</subscriptionRecoveryPolicy>
<pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit="1000"/>
</pendingMessageLimitStrategy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>

<!--
The managementContext is used to configure how ActiveMQ is exposed in
JMX. By default, ActiveMQ uses the MBean server that is started by
the JVM. For more information, see:

http://activemq.apache.org/jmx.html
-->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>

<!--
Configure message persistence for the broker. The default persistence
mechanism is the KahaDB store (identified by the kahaDB tag).
For more information, see:

http://activemq.apache.org/persistence.html
-->
<persistenceAdapter>
<kahaDB directory="E:/activemq_data"
enableJournalDiskSyncs="false"
indexWriteBatchSize="10000"
indexCacheSize="1000"
journalMaxFileLength="32mb"
/>
</persistenceAdapter>

<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage percentOfJvmHeap="70" />
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="50 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>

<!--
The transport connectors expose ActiveMQ over a given protocol to
clients and other brokers. For more information, see:

http://activemq.apache.org/configuring-transports.html
-->
<transportConnectors>

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

<shutdownHooks>
<bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
</shutdownHooks>

</broker>

<!--
Enable web consoles, REST and Ajax APIs and demos
The web consoles requires by default login, you can disable this in the jetty.xml file

Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
-->
<import resource="jetty.xml"/>

</beans>

接下来修改activemq2/conf/activemq.xml中<persistenceAdapter>标签下的<kahaDB>文件路径<directory>，修改为E:/activemq_data,详情如下：

http://www.apache.org/licenses/LICENSE-2.0

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" useJmx="true">

http://activemq.apache.org/slow-consumer-handling.html

<!--
The managementContext is used to configure how ActiveMQ is exposed in
JMX. By default, ActiveMQ uses the MBean server that is started by
the JVM. For more information, see:

http://activemq.apache.org/jmx.html
-->
<managementContext>
<managementContext createConnector="false"/>
</managementContext>

<!--
Configure message persistence for the broker. The default persistence
mechanism is the KahaDB store (identified by the kahaDB tag).
For more information, see:

<!--
The transport connectors expose ActiveMQ over a given protocol to
clients and other brokers. For more information, see:

http://activemq.apache.org/configuring-transports.html
-->
<transportConnectors>

<transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

</broker>

<!--
Enable web consoles, REST and Ajax APIs and demos
The web consoles requires by default login, you can disable this in the jetty.xml file

Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
-->
<import resource="jetty.xml"/>

</beans>

修改activemq2/conf/jetty.xml中端口号port

详情如下：

<!--
Licensed to the Apache Software Foundation (ASF) under one or more contributor
license agreements. See the NOTICE file distributed with this work for additional
information regarding copyright ownership. The ASF licenses this file to You under
the Apache License, Version 2.0 (the "License"); you may not use this file except in
compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or
agreed to in writing, software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied. See the License for the specific language governing permissions and
limitations under the License.
-->

<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

<bean id="securityConstraint" class="org.eclipse.jetty.util.security.Constraint">
<property name="name" value="BASIC" />
<property name="roles" value="user,admin" />

<property name="authenticate" value="true" />
</bean>
<bean id="adminSecurityConstraint" class="org.eclipse.jetty.util.security.Constraint">
<property name="name" value="BASIC" />
<property name="roles" value="admin" />

<property name="authenticate" value="true" />
</bean>
<bean id="securityConstraintMapping" class="org.eclipse.jetty.security.ConstraintMapping">
<property name="constraint" ref="securityConstraint" />
<property name="pathSpec" value="/admin/*,*.jsp" />
</bean>
<bean id="adminSecurityConstraintMapping" class="org.eclipse.jetty.security.ConstraintMapping">
<property name="constraint" ref="adminSecurityConstraint" />
<property name="pathSpec" value="*.action" />
</bean>
<bean id="securityHandler" class="org.eclipse.jetty.security.ConstraintSecurityHandler">
<property name="loginService" ref="securityLoginService" />
<property name="authenticator">
<bean class="org.eclipse.jetty.security.authentication.BasicAuthenticator" />
</property>
<property name="constraintMappings">
<list>
<ref bean="adminSecurityConstraintMapping" />
<ref bean="securityConstraintMapping" />
</list>
</property>
<property name="handler">
<bean id="sec" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
<list>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/hawtio" />
<property name="war" value="${activemq.home}/webapps/hawtio" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/admin" />
<property name="resourceBase" value="${activemq.home}/webapps/admin" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/fileserver" />
<property name="resourceBase" value="${activemq.home}/webapps/fileserver" />
<property name="logUrlOnStart" value="true" />
<property name="parentLoaderPriority" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/api" />
<property name="resourceBase" value="${activemq.home}/webapps/api" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.server.handler.ResourceHandler">
<property name="directoriesListed" value="false" />
<property name="welcomeFiles">
<list>
<value>index.html</value>
</list>
</property>
<property name="resourceBase" value="${activemq.home}/webapps/" />
</bean>
<bean id="defaultHandler" class="org.eclipse.jetty.server.handler.DefaultHandler">
<property name="serveIcon" value="false" />
</bean>
</list>
</property>
</bean>
</property>
</bean>

</bean>

</beans>

修改完成后进入cmd，启动E:activemq1inactivemq.bat；E:activemq2inactivemq.bat，下面贴出测试代码：

package activemq.test;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;

public class ActivemqTest {

@Test
public void testQueueSender() {
try {
//通过username,password,url创建连接工厂接口
ConnectionFactory factory = new ActiveMQConnectionFactory(
"system",
"manager", "failover:(tcp://localhost:61616,tcp://localhost:61617)?initialReconnectDelay=100");
//通过连接工厂创建一个新的连接接口
Connection connection = factory.createConnection();
//通过连接接口创建一个会话接口消息应答模式：Session.AUTO_ACKNOWLEDGE
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//会话接口创建有关主题的目标接口
Destination destination = session.createQueue("test_queue");
//会话接口再根据目标接口来创建一个消息生产者接口
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
//调用会话生成一个文本消息
Message message = session.createTextMessage("发送消息：Hello ActiveMQ Text Messagedwq ！");
long time = System.currentTimeMillis();
//通过生产者接口Send将消息发布到ActiveMQ服务器
producer.send(message);
System.out.println((System.currentTimeMillis()-time) + "ms");
//关闭会话
session.close();
//关闭连接
connection.close();

} catch (Exception e) {
e.printStackTrace();
}
}

@Test
public void testQueueReceiver() {
try {
// 通过username,password,url创建连接工厂接口
ConnectionFactory factory = new ActiveMQConnectionFactory(
"system",
"manager", "failover:(tcp://localhost:61616,tcp://localhost:61617)?initialReconnectDelay=100");
// 通过连接工厂创建一个新的连接接口
Connection connection = factory.createConnection();
connection.start();
// 通过连接接口创建一个会话接口
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
// 会话接口创建有关主题的目标接口
Destination destination = session.createQueue("test_queue");
// 会话接口再根据目标接口来创建一个消息消费者接口
MessageConsumer consumer = session.createConsumer(destination);
long time = System.currentTimeMillis();
// 消息消费者接收消息同步消息
TextMessage message = (TextMessage) consumer.receive();
if (null != message) {
System.out.println("收到消息：" + message.getText());
}
System.out.println((System.currentTimeMillis()-time) + "ms");
// 关闭会话
session.close();

} catch (Exception e) {
e.printStackTrace();
}
}
}

linux环境与上面基本一致，只需要更改文件地址即可。