多种方式配置ActiveMQ集群

野性酷女 2022-05-14 04:35 200阅读 0赞

# ActiveMQ静态网络配置 #

1，环境准备

\* 三台Linux服务器（虚拟机即可）

\* ActiveMQ安装包

\* 关闭服务器，解压安装包文件到服务器；依次启动，查看服务是否正常；一切就绪后进行文件配置

![70][]

2，配置文件

\* 在activemq.xml中修改broker名称

\* 在activemq.xml文件中配置静态网络的集群节点

\* 三台服务器中全部配置，保证三台服务器间可以正常互通

\* 三台启动完成后，页面效果

![70 1][]

3，静态网络互通测试

\* 服务端发送消息到128节点

public class ActiveMQProducer {
        public static void main(String[] args) throws JMSException {
            // 获取链接工厂
            ConnectionFactory connectionFactory =
                    new ActiveMQConnectionFactory("tcp://192.168.91.128:61616");
            // 创建链接
            Connection connection = connectionFactory.createConnection();
            connection.start();
            // 创建会话
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建目的地
            Destination destination = session.createQueue("myQueue");
            // 创建生产者
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            // 创建消息
            for (int i = 1; i <= 10; i++) {
                TextMessage message = session.createTextMessage("producer send message : " + i);
                // 生产者发送消息
                producer.send(message);
            }
            session.commit();
            session.close();
            System.out.println("消息发送成功");
        }
    }

![70 2][]

\* 消费端1从129节点消费消息

public class ActiveMQConsumer {
        public static void main(String[] args) throws JMSException, InterruptedException {
            ConnectionFactory connectionFactory = 
                    new ActiveMQConnectionFactory("tcp://192.168.91.129:61616?" +
                            "jms.optimizeAcknowledge=true&" +
                            "jms.optimizeAcknowledgeTimeOut=10000");
            // 创建链接
            Connection connection = connectionFactory.createConnection();
            connection.start();
            // 创建会话
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建目的地
            Destination destination = session.createQueue("myQueue?consumer.prefetchSize=10");
            // 创建消费者
            MessageConsumer consumer = session.createConsumer(destination);
            for (int i = 0; i < 10; i++) {
                // 接收信息
                TextMessage textMessage = (TextMessage) consumer.receive();
                // 消费信息
                System.out.println("消费者1 : " + textMessage.getText());
                session.commit();
            }
    
        }
    }

![70 3][]

\* 128节点显示129节点消费

![70 4][]

\* 129节点消费记录

![70 5][]

\* 消费端2从130节点消费消息

public class ActiveMQConsumer1 {
        public static void main(String[] args) throws JMSException, InterruptedException {
            ConnectionFactory connectionFactory =
                    new ActiveMQConnectionFactory("tcp://192.168.91.130:61616?" +
                            "jms.optimizeAcknowledge=true&" +
                            "jms.optimizeAcknowledgeTimeOut=10000");
            // 创建链接
            Connection connection = connectionFactory.createConnection();
            connection.start();
            // 创建会话
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建目的地
            Destination destination = session.createQueue("myQueue?consumer.prefetchSize=10");
            // 创建消费者
            MessageConsumer consumer = session.createConsumer(destination);
            while (true) {
                // 接收信息
                TextMessage textMessage = (TextMessage) consumer.receive();
                // 消费信息
                System.out.println("消费者2 : " + textMessage.getText());
                session.commit();
            }
    
        }
    }

![70 6][]

\* 128节点显示130节点消费

![70 7][]

\* 130节点消费记录

![70 8][]

4，消息回流

\* 静态网络配置好后，如果节点服务全部正常，则整个发布消费流程不会存在问题；但是如果在服务发布消费过程中，存在节点宕机，这就会存在消息丢失的风险，配置消息回流能够有效的防止消息丢失；

\* 配置方式 -- 在activemq.xml标签<policyEntries>中添加配置信息

<policyEntry queue=">" enableAudit="false"> 
    	<networkBridgeFilterFactory> 
    		<conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/> 
    	</networkBridgeFilterFactory> 
    </policyEntry>

\* 配置完成后重启服务使其生效

5，消息回流测试

\* 服务端发送消息到128节点

![70 9][]

\* 消费端消费部分消息后模拟宕机，重启activemq服务

![70 10][]

![70 11][]

\* 重启完成后，重新执行消费端进行消费，可以发现消费端继续进行消费，有部分重复是因为多次执行

![70 12][]

# 基于LevelDB+Zookeeper构建ActiveMQ集群 #

1，环境准备

\* 三台Linux服务器

\* ActiveMQ安装包

\* Zookeeper集群

2，配置文件 -- activemq.xml

\* 修改brokerName，三台机器brokerName统一修改为“activemq\_cluster”

\* 修改存储方式（默认为kahadb），修改为levelDB+ZK集群方式，hostName为当前节点IP

\* 集群启动：启动后只有当前主节点可以进行访问，且zookeeper节点上注册三个activemq服务节点

![70 13][]

![70 14][]

3，服务端发布消息

public class ActiveMQClusterProducer {
        public static void main(String[] args) throws JMSException {
            // 获取链接工厂
            ConnectionFactory connectionFactory =
                    new ActiveMQConnectionFactory(
                            "failover:(" +
                            "tcp://192.168.91.128:61616," +
                            "tcp://192.168.91.129:61616," +
                            "tcp://192.168.91.130:61616)?randomize=false");
            // 创建链接
            Connection connection = connectionFactory.createConnection();
            connection.start();
            // 创建会话
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建目的地
            Destination destination = session.createQueue("myQueue");
            // 创建生产者
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            // 创建消息
            for (int i = 1; i <= 10; i++) {
                TextMessage message = session.createTextMessage("producer send message : " + i);
                // 生产者发送消息
                producer.send(message);
            }
            session.commit();
            session.close();
            System.out.println("消息发送成功");
        }
    }

![70 15][]

4，客户端消费消息

public class ActiveMQConsumer {
        public static void main(String[] args) throws JMSException, InterruptedException {
            ConnectionFactory connectionFactory =
                    new ActiveMQConnectionFactory(
                            "failover:(" +
                                    "tcp://192.168.91.128:61616," +
                                    "tcp://192.168.91.129:61616," +
                                    "tcp://192.168.91.130:61616)?randomize=false");
            // 创建链接
            Connection connection = connectionFactory.createConnection();
            connection.start();
            // 创建会话
            Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建目的地
            Destination destination = session.createQueue("myQueue?consumer.prefetchSize=10");
            // 创建消费者
            MessageConsumer consumer = session.createConsumer(destination);
            for (int i = 0; i < 100; i++) {
                // 接收信息
                TextMessage textMessage = (TextMessage) consumer.receive();
                // 消费信息
                System.out.println("消费者1 : " + textMessage.getText());
                session.commit();
            }
    
        }
    }

![70 16][]

5，参数解析

\* randomize=true 客户端会随机对三个节点进行连接，当第一个节点宕机了，会依次链接其他节点

\* randomize=false , 所有的客户端连接都会打在第一个ip:port 上，只有第一个挂了，他才会打在第二个mq 上

# 基于kahadb构建ActiveMQ集群 #

1，环境准备

\* 三台Linux服务器

\* ActiveMQ安装包

2，构建挂载在三台服务器下的共享文件夹

\* 安装NFS服务

\# yum install nfs-utils

\* 服务端创建共享文件夹，修改/etc/exports文件

\# /home/work 192.168.91.\*(insecure,rw,sync,no\_root\_squash)

\* 启动NFS和rpcbind

\# service rpcbind start

\# service nfs start

\* 服务端检查服务器的NFS状态

\# showmount -e

\* 客户端查询服务端共享目录

\# showmount -e 192.168.91.128

\* 创建客户端共享目录

\# cd /mnt

\# mkdir static

\* 挂载服务端的共享目录

\# mount -t nfs -o nolock,nfsvers=3,vers=3 192.168.91.128:/home/work /mnt/static

\* 挂载在服务器下的共享文件夹为 /mnt/static

3，集群搭建

\* 配置broker信息，brokerName三个节点不能一样，dataDirectory为上一步配置的共享文件夹

\* 加载配置信息

\* 启动后集群状态

![70 17][]

4，集群测试

-- 部分配置存在问题，正在处理中

[70]: /images/20220514/2cc41061a1bf4620a5640410ad4ac9dc.png
[70 1]: /images/20220514/d64f51ddb2f34b7492cfd6a8843fc5dc.png
[70 2]: /images/20220514/3c8aa1c8e68f47a89829128bc41a4680.png
[70 3]: /images/20220514/7917b8e69f924d21af2567a06f0dda0b.png
[70 4]: /images/20220514/79265b07d49c4a2a8189f3c9fb740d2d.png
[70 5]: /images/20220514/0c779d0df8cf4e2b9688d59b873b77c8.png
[70 6]: /images/20220514/f1dc9989292144b59c4f5c4b865f544b.png
[70 7]: /images/20220514/75da76b7fad7424d9efb8f568f6b6a57.png
[70 8]: /images/20220514/e519488c2d9f4350881216c9b7a6ab7e.png
[70 9]: /images/20220514/b6a51b9f043b49ddbda6f7b21cbbed87.png
[70 10]: /images/20220514/9879e89590c74e3694dabb76af33ed54.png
[70 11]: /images/20220514/ab6d710b41bc4d3b99379f02aaf7fd13.png
[70 12]: /images/20220514/65d1d4035781476999500141a437207c.png
[70 13]: /images/20220514/89d5bf5e0ae646cb99d5f9ff746ff6d4.png
[70 14]: /images/20220514/a58df9e6cba942a48b3a53c406256cca.png
[70 15]: /images/20220514/e31341ba25cf418ea933f9052f51b8a0.png
[70 16]: /images/20220514/c30eebbe65424c608d6c527b6937865f.png
[70 17]: /images/20220514/6c2356a9794f42ed99d37c2ac20aee9b.png