0
点赞
收藏
分享

微信扫一扫

ActiveMQ - 集群

灯火南山 2022-05-23 阅读 61
ActiveMQ集群消费者生产者配置虚拟化云计算


ActiveMQ - 集群_集群 ActiveMQ - 集群_配置_02 ActiveMQ - 集群_生产者_03 ActiveMQ - 集群_生产者_04 ActiveMQ - 集群_消费者_05 ActiveMQ - 集群_ActiveMQ_06

ActiveMQ - 集群_集群_07 ActiveMQ - 集群_消费者_08 ActiveMQ - 集群_配置_09 ActiveMQ - 集群_集群_10 ActiveMQ - 集群_集群_10 ActiveMQ - 集群_配置_09 ActiveMQ - 集群_消费者_13 ActiveMQ - 集群_ActiveMQ_14

配置过程

1. 节点准备


mkdir activemq创建目录
cp -rf apache-activemq-5.15.0 activemq/activemq-a
cp -rf apache-activemq-5.15.0 activemq/activemq-b
cp -rf apache-activemq-5.15.0 activemq/activemq-c
cd activemq
mkdir kahadb


2. 配置a节点


cd activemq-a/
cd conf/
vim activemq.xml
    <networkConnectors>
              <networkConnector name="local_network" uri="static:(tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)" />
    </networkConnectors>
vim jetty.xml：配置管理端口号，a节点使用默认端口，无须配置


3. 配置b节点


vim activemq.xml
配置网络连接器
        <networkConnectors>
      <networkConnector name="network_a" uri="static:(tcp://127.0.0.1:61616)" />
    </networkConnectors>
配置持久化存储路径
    <persistenceAdapter>
        <kahaDB directory="/studio/activemq/kahadb"/>
    </persistenceAdapter>
配置服务端口
<transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
vim jetty.xml
配置管理端口号
<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
         <!-- the default port number for the web console -->
    <property name="host" value="0.0.0.0"/>
    <property name="port" value="8162"/>
</bean>


4. 配置c节点


vim activemq.xml
配置网络连接器
        <networkConnectors>
      <networkConnector name="network_a" uri="static:(tcp://127.0.0.1:61616)" />
    </networkConnectors>
配置持久化存储路径
    <persistenceAdapter>
        <kahaDB directory="/studio/activemq/kahadb"/>
    </persistenceAdapter>
配置服务端口
<transportConnector name="openwire" uri="tcp://0.0.0.0:61618?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
vim jetty.xml
配置管理端口号
<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
         <!-- the default port number for the web console -->
    <property name="host" value="0.0.0.0"/>
    <property name="port" value="8163"/>
</bean>


5.启动服务

回到activemq目录，分别启动a，b，c三个节点


./activemq-a/bin/activemq start
./activemq-b/bin/activemq start
./activemq-c /bin/activemq start


检查是否都启动成功


ps -ef |grep activemq


检查是否对外提供服务，即端口是否被监听（占用）


netstat -anp |grep 61616
netstat -anp |grep 61617
netstat -anp |grep 61618


检查发现61618即c节点没有提供服务，但是c节点的进程是启动成功了的。因为b节点和c点击是master/slave配置，现在b节点获取到了共享文件夹的所有权，所以c节点正在等待获得资源，并且提供服务。即c节点在未获得资源之前，是不提供服务的。

测试，把b节点杀掉，看c节点能不能提供61618的服务


./activemq-b/bin/activemq stop
netstat -anp |grep 61618
./activemq-b/bin/activemq start
netstat -anp |grep 61617


检查发现，重新启动b节点后，b节点61617端口并没有提供服务，是因为现在b节点成为了slave节点，而c节点成为了master节点。所以，现在b节点启动了，但是它并不对外提供服务。只有当c节点出现问题后，b节点才对外提供服务。

6.通过代码测试集群配置是否生效

生产者

package com.myimooc.jms.queue;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * App 生产者-队列模式-集群配置测试
 */
public class AppProducerTest {

    /** failover 为状态转移的存在部分
     * 因a节点只作为消费者使用，所以这里不配置61616节点了。
     * */
    private static final String URL = "failover:(tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)?randomize=true";

    /** 指定队列的名称 */
    private static final String QUEUE_NAME = "queue-test";
    
    public static void main(String[] args) throws JMSException {
        
        // 1.创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        
        // 2.创建Connection
        Connection connection = connectionFactory.createConnection();
        
        // 3.启动连接
        connection.start();
        
        // 4.创建会话（第一个参数：是否在事务中处理）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // 5. 创建一个目标
        Destination destination = session.createQueue(QUEUE_NAME);
        
        // 6.创建一个生产者
        MessageProducer producer = session.createProducer(destination);
        
        for (int i = 0; i < 100; i++) {
            
            // 7.创建消息
            TextMessage textMessage = session.createTextMessage("test" + i);
            
            // 8.发布消息
            producer.send(textMessage);
            
            System.out.println("消息发送：" + textMessage.getText());
        }
        
        // 9.关闭连接
        connection.close();
    }
    
}

消费者

package com.myimooc.jms.queue;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * App 消费者-队列模式-集群配置测试
 */
public class AppConsumerTest {

    /** failover 为状态转移的存在部分
     * */
    private static final String URL = "failover:(tcp://127.0.0.1:61616,tcp://127.0.0.1:61617,tcp://127.0.0.1:61618)?randomize=true";

    /** 指定队列的名称 */
    private static final String QUEUE_NAME = "queue-test";
    
    public static void main(String[] args) throws JMSException {
        
        // 1.创建ConnectionFactory
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
        
        // 2.创建Connection
        Connection connection = connectionFactory.createConnection();
        
        // 3.启动连接
        connection.start();
        
        // 4.创建会话（第一个参数：是否在事务中处理）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        
        // 5.创建一个目标
        Destination destination = session.createQueue(QUEUE_NAME);
        
        // 6.创建一个消费者
        MessageConsumer consumer = session.createConsumer(destination);
        
        // 7.创建一个监听器
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage)message;
                try {
                    System.out.println("接收消息：" + textMessage.getText());
                } catch (JMSException e) {
                    System.out.println("接收消息异常：");
                    e.printStackTrace();
                }
            }
        });
        
        // 8.关闭连接
        //connection.close();
    }
    
}

运行生产者，然后到管理界面查看消息发送到了那里


http://127.0.0.1:8161
http://127.0.0.1:8162
http://127.0.0.1:8163


查看发现，8162无法访问，是因为b节点是slave节点，不提供服务，消息都发送到了c节点

把8163即c节点宕掉后，运行消费者，查看消息是否能够使用


./activemq-c/bin/activemq stop



举报

相关推荐

ActiveMQ集群应用

数据库网络javaSystembc办公效率

ActiveMQ 集群部署

中间件

ActiveMQ 常见集群模式

apachexmljavaOpenStack云计算

ActiveMq伪集群设置

码哥java后端集...

基于zookeeper(集群)+LevelDB的ActiveMq高可用集群安装、配置、测试

ActiveMQ持久化xmlapache编程语言

activemq集群注册到zookeeper的访问路由

。Python后端开发

K8S如何部署ActiveMQ（单机、集群）

android

【ActiveMq】linux ActiveMq安装

apachelinuxjava

zookeeper+ActiveMQ集群启动后无法访问

java-zookeeperjava-activemqzookeeperapachejar虚拟化云计算

ActiveMQ - SpringJMS 之 ActiveMQ

ActiveMQSpringJMS 之 ActiveMQ主题模式队列模式springOpenStack云计算
0 条评论