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测试过程

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

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

消费者1接收消息情况：

消费者2接收消息情况：

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

3.4测试结论

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

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

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

1.基础功能测试

1.1点对点模式(Queue)

1.1.1引入依赖

1.1.2生产者发布Queue消息

1.1.3消费者接收Queue消息

1.1.4多消费者模式

1.2发布/订阅模式(Topic)

1.2.1生产者发布Topic消息

1.2.2消费者接收Topic消息

1.2.3多消费者模式

2.高可用测试

2.1测试方案一

2.1.1测试用例

2.1.2测试代码

2.1.3测试过程

2.2测试方案二

2.2.1测试用例

2.2.2测试代码

2.2.3测试过程

2.3测试结论

3.负载均衡测试

3.1测试用例

3.2测试代码

3.3测试过程

3.4测试结论

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

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

1.基础功能测试

1.1点对点模式(Queue)

1.1.1引入依赖

1.1.2生产者发布Queue消息

1.1.3消费者接收Queue消息

1.1.4多消费者模式

1.2发布/订阅模式(Topic)

1.2.1生产者发布Topic消息

1.2.2消费者接收Topic消息

1.2.3多消费者模式

2.高可用测试

2.1测试方案一

2.1.1测试用例

2.1.2测试代码

2.1.3测试过程

2.2测试方案二

2.2.1测试用例

2.2.2测试代码

2.2.3测试过程

2.3测试结论

3.负载均衡测试

3.1测试用例

3.2测试代码

3.3测试过程

3.4测试结论

热门文章

最新文章

相关课程

相关电子书

相关实验场景