ActiveMQ集群搭建好之后,接下来就该测试一下了。
一、先安装Zookeeper 的eclipse插件吧。
1. 打开 eclipse, Help -》 Install New Software,点击Add,添加地址: http://www.massedynamic.org/eclipse/updates/
2. 点击OK,勾选Zookeeper插件,另外两个貌似没用
3. 安装完之后,点击Window-》Show View-》Other,选择安装的Zookeeper插件
4. 出现Zookeeper控制台后,点击右上角的按钮,增加Zookeeper节点。
出现界面
点击Add
添加你的Zookeeper IP,Finish
5. 查看集群信息。
如果你的Zookeeper+ActiveMQ开启的话,会出现
二、代码
一个简单的生产者:
package mq.cluster; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class Sender { public static void main(String[] args) throws Exception { //1. 建立一个ConnectionFactory. 默认tcp://0.0.0.0:61616
String userName = ActiveMQConnectionFactory.DEFAULT_USER;
String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
String brokerURL = "failover:(tcp://192.168.127.129:61616,tcp://192.168.127.130:61616,tcp://192.168.127.131:61616)?Randomize=false";
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, password, brokerURL); //2. 通过ConnectionFactory建立一个Connection连接,并且调用start方法开启
Connection connection = connectionFactory.createConnection();
connection.start(); //3. 通过Connection创建Session,用于接收消息[第一个参数:是否启用事务;第二个参数:设置签收模式]
// Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);//手工签收--常用 //4. 通过Session创建Destination对象
Destination destination = session.createQueue("foodQueue"); //5. 通过Session创建发送或接受对象
MessageProducer messageProducer = session.createProducer(null); //7. 使用JMS规范里面消息类型之一 TextMessage 来创建数据,用MessageProducer来发送
for (int i = 1; i < 500; i++) {
TextMessage message = session.createTextMessage();
message.setText("大萝卜" + i);
//参数:目标,消息,传递数据的模式,优先级,消息的过期时间
messageProducer.send(destination, message, DeliveryMode.NON_PERSISTENT, 0, 1000*60);
//System.out.println("生产者:" + message.getText());
Thread.sleep(1000);
} //使用事务要手动提交
//session.commit(); //8. 关闭连接
connection.close(); }// main
}
因为是集群,所以brokerUrl要与众不同,
String brokerURL = "failover:(tcp://192.168.127.129:61616,tcp://192.168.127.130:61616,tcp://192.168.127.131:61616)?Randomize=false";
一个简单的消费者:
package mq.cluster; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class Receiver { public static void main(String[] args) throws Exception { //1. 建立一个ConnectionFactory. tcp://0.0.0.0:61616
String userName = ActiveMQConnectionFactory.DEFAULT_USER;
String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
String brokerURL = "failover:(tcp://192.168.127.129:61616,tcp://192.168.127.130:61616,tcp://192.168.127.131:61616)?Randomize=false";
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, password, brokerURL); //2. 通过ConnectionFactory建立一个Connection连接,并且调用start方法开启
Connection connection = connectionFactory.createConnection();
connection.start(); //3. 通过Connection创建Session,用于接收消息[第一个参数:是否启用事务;第二个参数:设置签收模式]
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //4. 通过Session创建Destination对象
Destination destination = session.createQueue("foodQueue"); //5. 通过Session创建发送或接受对象
MessageConsumer messageConsumer = session.createConsumer(destination); //7. 使用JMS规范里面消息类型之一 TextMessage 来创建数据
while (true) {
TextMessage message = (TextMessage) messageConsumer.receive();
if(message == null) break;
System.out.println("消费者:" + message.getText());
} //8. 关闭连接
connection.close(); }// main
}
先开启消费者,让它先阻塞,准备吃饭。然后开启生产者:
浏览器打开提供服务的MQ控制台
一切正常
三、关于负载均衡
画个图先~仅供参考【这部分真的仅供参考啊!!别喷我,但是可以告诉我怎样是对的,希望大家指正】
目前的架构:
负载均衡方案:=========================================
第二种方案呢,假设我们整个应用是个食堂,这个食堂里面有好多饭口,每个饭口有好多大爷大妈(咦~~~)。每个饭口实现了高可用,即使有大妈不上班,还有大爷顶上。针对好多学生(用户),实现了负载均衡,学生被基本均匀的分配到每一个饭口上。
食堂----应用
饭口----集群
大妈----服务器
==================================
现在我们利用Broker-Cluster 可以解实现负载均衡。配置的方法就是:
打开activemq.xml文件,在persistenceAdapter的前面加上
< networkConnectors >
</ networkConnectors >
|
就拿这两个集群来说:
集群1 >>>>>>>>链接>>>>>>>> 集群2:192.168.127.1,192.168.127.2,192.168.127.3 加上
<networkConnectors>
<networkConnector uri="static:(tcp://192.168.127.4:61616,tcp://192.168.127.5:61616,tcp://192.168.127.6:61616)" duplex="false"/>
</networkConnectors>
集群2 >>>>>>>>链接>>>>>>>> 集群1: 192.168.127.4,192.168.127.5,192.168.127.6 加上
<networkConnectors>
<networkConnector uri="static:(tcp://192.168.127.1:61616,tcp://192.168.127.2:61616,tcp://192.168.127.3:61616)" duplex="false"/>
</networkConnectors>
说白了就是把对方的地址拿过来
如果是三个集群呢?
那么就多加一个<networkConnector uri="tcp://..."/>
比如多了一个集群3(IP:192.168.127.7,192.168.127.8,192.168.127.9)
集群1 >>>>>>>>链接>>>>>>>> 集群2&集群3:192.168.127.1,192.168.127.2,192.168.127.3 加上
<networkConnectors>
<networkConnector uri="static:(tcp://192.168.127.4:61616,tcp://192.168.127.5:61616,tcp://192.168.127.6:61616)" duplex="false"/>
<networkConnector uri="static:(tcp://192.168.127.7:61616,tcp://192.168.127.8:61616,tcp://192.168.127.9:61616)" duplex="false"/>
</networkConnectors>
集群2 >>>>>>>>链接>>>>>>>> 集群1&集群3 和 集群3 >>>>>>>>链接>>>>>>>> 集群1&集群2 照葫芦画瓢
最后还是要说一句:仅供参考啊!!!!!