ActiveMQ高可用+负载均衡集群之功能测试

1.基础功能测试

ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。JMS的全称是Java Message Service，即Java消息服务。它主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息，而消费者负责接收消息。而消息的传递有两种类型，主要如下：

• 一种是点对点的，即一个生产者和一个消费者一一对应。
  
• 另一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。
  

ActiveMQ和JMS的消息类型对应如下

JMS消息模型	P2P消息模型	Pub/Sub消息模型
ActiveMQ	Queue队列	Topic队列
特点	一对一，生产者生产了一个消息，只能由一个消费者进行消费	一对多，生产者生产了一个消息，可以由多个消费者进行消费

接下来将对两种类型的场景进行分别验证。

1.1点对点模式(Queue)

点对点的模式主要建立在某个queue上，消息可以被同步或异步的发送和接收。点对点的消息模式可以有多个发送端，多个接收端，但是每个消息只会给一个Consumer传送一次。

1.1.1引入依赖

1.ActiveMQ依赖

<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>5.15.12</version> </dependency>

2.Springboot-ActiveMQ连接池依赖
如果要启用连接池,且使用springboot2.0+及以下版本的时候，maven配置依赖是：

<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool</artifactId> </dependency>

如果要启用连接池,且使用springboot2.1+的时候，maven配置依赖是：

<dependency> <groupId>org.messaginghub</groupId> <artifactId>pooled-jms</artifactId> </dependency>

1.1.2生产者发布Queue消息

由于ActiveMQ的客户端只能访问Master的Broker，其他处于Slave的Broker不能被访问，所以客户端连接Broker应该使用failover协议，具体代码如下。

String url="failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)";//服务地址 String queueName = "queue-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 //第一个参数：是否开启事务 //第二个参数：消息是否自动确认 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个队列 Destination destination = session.createQueue(queueName); //6.创建一个生产者 MessageProducer producer = session.createProducer(destination); for (int i = 0; i < 20; i++) { //7.创建消息 TextMessage textMessage = session.createTextMessage("我是Queue消息生产者：" + i); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } //8.发送消息 producer.send(textMessage); System.out.println("发送第一组消息：" + i); } connection.close();

运行代码后，可在ActiveMQ控制台查看对应的Queue信息，此时有消息待接收。

1.1.3消费者接收Queue消息

Consumer 可以使用 MessageConsumer.receive() 同步地接收消息，也可以通过使用MessageConsumer.setMessageListener() 注册一个 MessageListener 实现异步接收。
这里使用异步接收的方式消费消息，具体代码如下：

String url="failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)";//服务地址 String queueName="queue-testqq";//要消费的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory=new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection= factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 /** 第一个参数，是否使用事务 如果设置true，操作消息队列后，必须使用 session.commit(); 如果设置false，操作消息队列后，不使用session.commit(); */ Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination=session.createQueue(queueName); //6.创建一个消费者 MessageConsumer consumer=session.createConsumer(destination); //7.创建一个监听器 consumer.setMessageListener(new MessageListener() { public void onMessage(Message arg0) { TextMessage textMessage=(TextMessage)arg0; try { System.out.println("接收消息："+textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } });

执行代码后，可以在控制台如下输出：

在ActiveMQ控制台查看对应的Queue信息，此时消息已被消费。

1.1.4多消费者模式

同时开启2个以上的消费者，再次运行生产者，观察每个消费者控制台的输出。

观察后，得出结论：一条消息只会被一个消费者会接收消费，不可重复消费。同时还发现，多个消费者的情况下消息会被均分，即负载均衡策略。
生产者发送消息情况：

消费者1接收消息情况：

消费者2接收消息情况：

1.2发布/订阅模式(Topic)

Pub/Sub（发布/订阅，Publish/Subscribe）消息域使用topic作为Destination，发布者向topic发送消息，订阅者注册接收来自topic的消息。发送到topic的任何消息都将自动传递给所有订阅者。接收方式（同步和异步）与P2P域相同。

1.2.1生产者发布Topic消息

Pub/Sub模式与P2P模式在代码实现层面基本一样，变化的只有Queue与Topic。生产者发布Topic消息的具体代码如下：

String url="failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)";//服务地址，端口默认61616 String topicName="topic-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory=new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection= factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 (参数1：是否启动事务,参数2：消息确认模式) Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination=session.createTopic(topicName); // Topic topic = session.createTopic(topicName); //6.创建一个生产者 MessageProducer producer=session.createProducer(destination); for (int i = 0; i < 15; i++) { //7.创建消息 TextMessage textMessage=session.createTextMessage("我是topic类型消息生产者："+i); //8.发送消息 producer.send(textMessage); System.out.println("发送消息："+i); } connection.close();

此时，消息生产者先不启动。Pub/Sub模式下必须先启动sub，否则在启动sub之前发布的消息是不能消费的，就像你今天开始订报纸，那今天之前的报纸你肯定是收不到了，发布/订阅模式与此同理。

1.2.2消费者接收Topic消息

与P2P模式相同，Pub/Sub模式的消息接收方式也有两种：同步接收和异步接收。这里采用异步接收的方式消费Topic消息，具体代码如下：

String url = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)";//服务地址，端口默认61616 String topicName = "topic-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination = session.createTopic(topicName); //6.创建一个消费者 MessageConsumer consumer = session.createConsumer(destination); //7.接收消息，可选择同步接收或者异步接收 /*//消费者同步接收，receive(long timeout)主线程阻塞式等待下一个消息到来，可设置超时时间，超时则返回null。 TextMessage message = (TextMessage) consumer.receive(1000); System.out.println("同步接收Topic消息: " + message);*/ //消费者异步接收，创建一个监听器 consumer.setMessageListener(new MessageListener() { public void onMessage(Message arg0) { TextMessage textMessage = (TextMessage) arg0; try { System.out.println("异步接收Topic消息：" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } });

运行代码，可以看到控制台输出已消费Topic消息。

ActiveMQ控制台也可以看到对应的Topic发布、订阅信息。如下

1.2.3多消费者模式

同时开启2个以上的消费者，再次运行生产者，观察每个消费者控制台的输出.

两个消费者运行情况如下：

可以发现：

• topic发布/订阅模式，一个消息可以被多个消费者消费
• topic发布/订阅模式要求消费者必须即时消费，即生产者发布消息时，消费者必须同时在线才可接收消费消息。

2.高可用测试

2.1测试方案一

2.1.1测试用例

1.生产者连接集群发送50条消息并设置每发送一条消息，sleep1秒
2.观察生产者发送消息所连接的节点，并将所在的节点停掉
3.观察生产者发送消息日志，查看所有消息是不是正常发送

2.1.2测试代码

1.生产者发布消息

String url = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)"; String queueName = "queue-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 //第一个参数：是否开启事务 //第二个参数：消息是否自动确认 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination = session.createQueue(queueName); // Queue queue = session.createQueue("test-Queue"); //6.创建一个生产者 MessageProducer producer = session.createProducer(destination); for (int i = 0; i < 50; i++) { //7.创建消息 TextMessage textMessage = session.createTextMessage("我是第一组消息生产者：" + i); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } //8.发送消息 producer.send(textMessage); System.out.println("发送第一组消息：" + i); } connection.close();

2.1.3测试过程

1.运行生产者发布消息，观察生产者控制台

2.停止mq1(61616)节点

3.继续观察生产者控制台
（1）此时61616节点已无法连接

（2）生产者已成功连接61619节点，并继续发送消息

2.2测试方案二

2.2.1测试用例

1.生产者连接集群发送20条消息
2.观察生产者发送消息所连接的节点，并将所在的节点停掉
3.消费者连接集群消费消息，观察消息消费情况

2.2.2测试代码

1.生产者发布消息

String url = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)"; String queueName = "queue-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 //第一个参数：是否开启事务 //第二个参数：消息是否自动确认 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination = session.createQueue(queueName); // Queue queue = session.createQueue("test-Queue"); //6.创建一个生产者 MessageProducer producer = session.createProducer(destination); for (int i = 0; i < 20; i++) { //7.创建消息 TextMessage textMessage = session.createTextMessage("我是第一组消息生产者：" + i); //8.发送消息 producer.send(textMessage); System.out.println("发送第一组消息：" + i); } connection.close();

2.消费者接收消息

String url = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)"; String queueName = "queue-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 /** 第一个参数，是否使用事务 如果设置true，操作消息队列后，必须使用 session.commit(); 如果设置false，操作消息队列后，不使用session.commit(); */ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination = session.createQueue(queueName); //6.创建一个消费者 MessageConsumer consumer = session.createConsumer(destination); //7.创建一个监听器 consumer.setMessageListener(new MessageListener() { public void onMessage(Message arg0) { TextMessage textMessage = (TextMessage) arg0; try { System.out.println("接收消息：" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } });

2.2.3测试过程

1.运行生产者发布消息，观察生产者控制台

2.停止mq4(61619)节点

3.启动消费者，观察消费情况

观察发现，消费者连接61618节点，并成功接收消费消息。

2.3测试结论

经测试，当前集群在启动消息生产者发送消息时，使生产者所在节点宕机的情况下，得出如下结论：
1.高可用架构的ActiveMQ集群，在生产消息的过程中生产者所在节点挂掉，客户端会暂时阻塞无法发送消息，但整体可用性不受影响。
2.高可用架构的ActiveMQ集群，在消息生产者所在节点挂掉后，消费者仍可正常消费消息
3.当前ActiveMQ集群若其中一个节点挂掉，ActiveMQ正常提供服务，不影响服务可用性

3.负载均衡测试

最终的架构就是两个master-slave集群相互连通,两个集群可以相互消费对方的消息,但是如果客户端所连接的集群挂掉客户端依然是不能发送消息的,也就是说activemq的负载均衡只是做到消费的负载均衡,高可用是靠master-slave来保证的。

3.1测试用例

1.启动两个消费者监听相同的queue，且服务地址均配置集群所有节点
2.生产者连接集群向指定的queue连续发送20条消息
3.观察两个生产者消费消息的日志

3.2测试代码

1.生产者发布消息

String url = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618,tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)"; String queueName = "queue-testqq";//要创建的消息名 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 //第一个参数：是否开启事务 //第二个参数：消息是否自动确认 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination = session.createQueue(queueName); //Queue queue = session.createQueue("test-Queue"); //6.创建一个生产者 MessageProducer producer = session.createProducer(destination); for (int i = 0; i < 20; i++) { //7.创建消息 TextMessage textMessage = session.createTextMessage("我是第一组消息生产者：" + i); //8.发送消息 producer.send(textMessage); System.out.println("发送第一组消息：" + i); } connection.close();

2.消费者接收消费消息

String url = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618," + "tcp://127.0.0.1:61619,tcp://127.0.0.1:61620,tcp://127.0.0.1:61621)"; String queueName = "queue-testqq";//要创建的消息名称 //1.创建ConnectiongFactory,绑定地址 ConnectionFactory factory = new ActiveMQConnectionFactory(url); //2.创建Connection Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话 /** 第一个参数，是否使用事务 如果设置true，操作消息队列后，必须使用 session.commit(); 如果设置false，操作消息队列后，不使用session.commit(); */ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //5.创建一个目标 Destination destination = session.createQueue(queueName); //6.创建一个消费者 MessageConsumer consumer = session.createConsumer(destination); //7.创建一个监听器 consumer.setMessageListener(new MessageListener() { public void onMessage(Message arg0) { TextMessage textMessage = (TextMessage) arg0; try { /* if (textMessage.getText().contains("10")) { System.out.println("======消息异常======="); throw new Exception(); }*/ System.out.println("接收消息：" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } });

3.3测试过程

1. 同时开启2个消费者，再运行生产者，观察每个消费者控制台的输出。
   

1. 再运行生产者，观察每个消费者控制台的输出。
   生产者发布消息情况：

消费者1接收消息情况：

消费者2接收消息情况：

观察发现，多个消费者的情况下消息会被均分，即负载均衡策略。且同一条消息只会被一个消费者会接收消费。

3.4测试结论

经测试，当前集群在多个消费者消费相同队列的情况下，可以实现消息消费的负载均衡，从而实现ActiveMQ集群的分流，提高集群吞吐率。

到这里，ActiveMQ集群的功能测试、高可用测试及负载均衡测试已完成，当前ActiveMQ集群高可用+负载均衡功能正常。