0
点赞
收藏
分享

微信扫一扫

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



举报

相关推荐

0 条评论