Broker-Cluster实现负载均衡(分流、提高吞吐率)

一般中间件都提供了横向扩展和纵向扩展，横向扩展就是我们经常说的负载均衡,纵向扩展提供了Master-Slaver;

消息中间件解决了应用之间的消息传递、解耦、异步的问题。

Broker-Cluster部署方式中，各个broker通过网络互相连接，并共享queue，提供了2种部署方式：static Broker-Cluster和Dynamic Broker-Cluster

1).static Broker-Cluster

将ActiveMq拷贝2份，分别命名:apache-activemq-5.10.0_M1,apache-activemq-5.10.0_M2， 下面就是配置activemq.xml:

M1做如下配置:(黑色是必须的，红色是可选的)

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="AS" dataDirectory="${activemq.data}"> <!--networkConnectors节点的配置必须要在persistenceAdapter节点之前 --> 

duplex:网络是否为双向的，默认为false。如果为true，则remote Broker不仅是“订阅者”，还可以作为消息的“生产者”，即remote Broker也会向Local Broker转发消息。 我们通常是，每个broker均与其他broker建立单向的networkConnector，比如有三个broker A/B/C，那么在A上配置2个单向的：A->B，A->C，那么在B上配置：B->A，B->C，以此论推。

<networkConnectors> <networkConnector uri="static:(tcp://192.168.1.9:61617)" duplex="true"/> </networkConnectors> <persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds" useDatabaseLock="false" /> </persistenceAdapter> <transportConnectors> <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors> <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql://192.168.1.4:3306/AS?relaxAutoCommit=true"/> <property name="username" value="root"/> <property name="password" value="root"/> <property name="poolPreparedStatements" value="true"/> </bean>

<broker>标签之外配置数据源mysql-ds

M2做如下配置(黑色是必须的，红色是可选的)

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="BS" dataDirectory="${activemq.data}"> <!--networkConnectors节点的配置必须要在persistenceAdapter节点之前 -->  <networkConnectors> <networkConnector uri="static:(tcp://192.168.1.9:61616)" duplex="true"/> </networkConnectors> <persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds" useDatabaseLock="false"/> </persistenceAdapter> <transportConnectors> <transportConnector name="openwire" uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors> <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql://192.168.1.4:3306/BS?relaxAutoCommit=true"/> <property name="username" value="root"/> <property name="password" value="root"/> <property name="poolPreparedStatements" value="true"/> </bean>

<broker>标签之外配置数据源mysql-ds

通过以上配置使M1和M2这两个 broker通过网络互相连接，并共享queue，启动M1和M2,可以看到如下启动日志：network connection has been established

Network connection between vm://AS#44 and tcp:///192.168.1.9:61617@45131 (BS) has been established

测试

Send发送类：

import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Sender { public static void main(String[] args) { ConnectionFactory connectionFactory; Connection connection = null; Session session; Destination destination; MessageProducer producer; connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://192.168.1.9:61616"); //connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://192.168.1.9:61618"); try { connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); destination = session.createQueue("FirstQueue"); producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.PERSISTENT); sendMessage(session, producer); session.commit(); } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) {} } } private static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 1; i <= 5; i++) { TextMessage message = session.createTextMessage("===发送的消息" + i); System.out.println("发送消息：" + "ActiveMq 发送的消息" + i); producer.send(message); } } }

Receiver接收类：

import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Receiver { public static void main(String[] args) { ConnectionFactory connectionFactory; Connection connection = null; Session session; Destination destination; MessageConsumer consumer; connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://192.168.1.9:61617"); //connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://192.168.1.9:61619"); try { connection = connectionFactory.createConnection(); connection.start(); session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); destination = session.createQueue("FirstQueue"); consumer = session.createConsumer(destination); while (true) { TextMessage message = (TextMessage) consumer.receive(1000); if (null != message) { System.out.println("收到消息" + message.getText()); } else { break; } } } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) {} } } }经测试Receiver可以接受到数据，表示M1和M2已经共享了queue

2).Dynamic Broker-Cluster

Dynamic Discovery集群方式在配置ActiveMQ实例时，不需要知道所有其它实例的URI地址

组播（multicast）基于UDP协议，它是指在一个可连通的网络中，某一个数据报发送源向一组数据报接收目标进行操作的过程。在这个过程中，数据报发送者只需要向这个组播地址（一个D类IP）发送一个数据报，那么加入这个组播地址的所有接收者都可以收到这个数据报。组播实现了网络中单点到多点的高效数据传送，能够节约大量网络带宽，降低网络负载。

对activemq.xml做如下配置: M1做如下配置:(黑色是必须的，红色是可选的)

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="AD" dataDirectory="${activemq.data}">

<networkConnectors> <networkConnector uri="multicast://239.0.0.5" duplex="false"/> </networkConnectors>

<!-- networkConnectors节点的配置必须要在persistenceAdapter节点之前 -->

<persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds" useDatabaseLock="false"/> </persistenceAdapter>

<transportConnectors> <transportConnector discoveryUri="multicast://239.0.0.5" name="openwire" uri="tcp://0.0.0.0:61618?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors> 配置JDBC方式的持久化适配器

<broker>标签之外配置数据源mysql-ds

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql://192.168.1.4:3306/AD?relaxAutoCommit=true"/> <property name="username" value="root"/> <property name="password" value="root"/> <property name="poolPreparedStatements" value="true"/> </bean>

M2做如下配置:(黑色是必须的，红色是可选的)

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="BD" dataDirectory="${activemq.data}"> <!--networkConnectors节点的配置必须要在persistenceAdapter节点之前 --> 

<networkConnectors> <networkConnector uri="multicast://239.0.0.5" duplex="false"/> </networkConnectors> <persistenceAdapter> <jdbcPersistenceAdapter dataSource="#mysql-ds" useDatabaseLock="false"/> </persistenceAdapter> <transportConnectors> <transportConnector discoveryUri="multicast://239.0.0.5" name="openwire" uri="tcp://0.0.0.0:61619?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors> 配置JDBC方式的持久化适配器

<broker>标签之外配置数据源mysql-ds

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql://192.168.1.4:3306/BD?relaxAutoCommit=true"/> <property name="username" value="root"/> <property name="password" value="root"/> <property name="poolPreparedStatements" value="true"/> </bean>

启动M1和M2,可以看到如下启动日志： network connection has been established

Network connection between vm://BD#0 and tcp:///192.168.1.9:61618@50410 (AD) has been established.

测试同static broker-cluster,可以得到相同的结果。 官网配置说明：http://activemq.apache.org/networks-of-brokers.html

Master-Slaver保证了数据的可靠性，Broker-Cluster提供了负载均衡，所以一般正式环境中都会采用：Master-Slaver+Broker-Cluster的模式

注意问题：

1.JMS之ActiveMQ启动时报锁定数据库的问题解决(JDBC方式时候)：activeMQ的broker在启动时会锁定数据库。我们每个人在调试时，自己的运行环境中就会运行一个broker，所以会出现争用锁的现象（如果只有一个人运行则不会出现这样的问题），报错如下：Cannot execute statement: impossible to write to binary log since BINLOG_FORMAT = STATEMENT and at least one table uses a storage engine limited to row-based logging. InnoDB is limited to row-logging when transaction isolation level is READ COMMITTED or READ UNCOMMITTED. SQLState: HY000 Vendor code: 1665 | org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter

处理办法：在activemq.xml中，修改jdbcPersistenceAdapter选项，添加一个：useDatabaseLock="false"

2. mysql JDBC驱动包mysql-connector-java-5.1.39.jar, commons-dbcp-1.4.jar,  commons-pool-1.6.jar   放到${ACTIVEMQ_HOME}/lib/下

3.错误如下：impossible to write to binary log since BINLOG_FORMAT = STATEMENT and at least one table uses a storage engine limited to row-based logging无法写入二进制日志自binlog_format 语句。

处理方法:修改my.cnf，binlog_format=mixed

4.端口61616,61617,8161等加入防火墙

5.chmod -R  777  /yxue/activeMQ  ,否则-bash: ./activemq: Permission denied