ActiveMQ学习-Network connectors（1）

http://blog.csdn.net/xiajun07061225/article/details/47068451

本文为network connectors的static connector学习笔记。

Network connectors

broker网络能够创建多个相互连接的ActiveMq实例组成的簇，以应对更加复杂的消息场景。Network connectors提供了broker之间的通信。 默认情况下，network connector是单向通道，它只会把收到的消息投递给与之建立连接的另一个broker。这通常称为forwarding bridge。ActiveMQ也支持双向通道，即duplex connector。下图是一个包含了这两者的复杂网络。

Network connector的XML配置如下：

[html] view plain copy

print?

1. <networkConnectors>
2. <networkConnector name="default-nc" uri="multicast://default"/>
3. </networkConnectors>

一个重要的概念-discovery discovery：是一个检测远程broker服务的进程。client通常需要感知所有的broker。broker，需要感知其他存在的broker，以建立broker的网络。 当我们想配置一个broker网络时，首要问题是：我们知道每个broker的准确地址吗？如果是，可以以静态的方式配置，将客户端连接到提前定义好的broker URI，这在你想完全控制所有资源的生产环境中比较常见。 如果客户端以及broker相互不知道彼此的地址，那么必须使用一种discovery机制来发现已有的broker。这种设置在开发环境下比较常见，易于配置和维护。

static network

概念介绍

只要我们知道了想要使用的broker的地址，就可以使用static配置方式。

Static connector

用来创建网络中多个broker的静态配置。协议使用组合URI，即URI中包含其他URI。格式如下： static:(uri1,uri2,uri3,...) ?key=value XML中配置示例：

[html] view plain copy

print?

1. <networkConnectors>
2. <networkConnector name="local network"
3. uri="static://(tcp://remotehost1:61616,tcp://remotehost2:61616)"/>
4. </networkConnectors>

程序实例

为了更好的理解，可以通过一个发布者-订阅者的例子来进行说明。（demo来自ActiveMQ in action上的例子） 这个例子使用下图所示的broker拓扑结构：

BrokerA与brokerB单向相连，当生产者把消息发送给brokerA时，他们会被投递给有订阅需求的broker。这个时候，会被brokerA投递给brokerB。 详细代码如下。 brokerB配置（brokerB.xml）：

[html] view plain copy

print?

1. <beans
2. xmlns="http://www.springframework.org/schema/beans"
3. xmlns:amq="http://activemq.org/config/1.0"
4. xmlns:broker="http://activemq.apache.org/schema/core"
5. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
6. xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd  
7.   http://activemq.org/config/1.0 http://activemq.apache.org/schema/activemq-core.xsd  
8.   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.2.0.xsd  
9.   http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
11. <!-- Allows us to use system properties as variables in this configuration file -->
12. <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
14. <!-- 定义一个broker -->
15. <broker xmlns="http://activemq.apache.org/schema/core" brokerName="BrokerB" dataDirectory="${activemq.base}/data">
17. <!-- The transport connectors ActiveMQ will listen to -->
18. <transportConnectors>
19. <transportConnector name="openwire" uri="tcp://localhost:61617" />
20. </transportConnectors>
22. </broker>
24. </beans>

brokerA配置（brokerA.xml）：

[html] view plain copy

print?

1. <beans
2. xmlns="http://www.springframework.org/schema/beans"
3. xmlns:amq="http://activemq.org/config/1.0"
4. xmlns:broker="http://activemq.apache.org/schema/core"
5. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
6. xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd  
7.   http://activemq.org/config/1.0 http://activemq.apache.org/schema/activemq-core.xsd  
8.   http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.2.0.xsd  
9.   http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
11. <!-- Allows us to use system properties as variables in this configuration file -->
12. <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
14. <broker xmlns="http://activemq.apache.org/schema/core" brokerName="BrokerA" dataDirectory="${activemq.base}/data">
16. <!-- The transport connectors ActiveMQ will listen to -->
17. <transportConnectors>
18. <transportConnector name="openwire" uri="tcp://localhost:61616" />
19. </transportConnectors>
21. <!-- 定义一个network连接器，连接到其他broker -->
22. <networkConnectors>
23. <networkConnector uri="static:(tcp://localhost:61617)" />
24. </networkConnectors>
26. </broker>
28. </beans>

消息生产者（Publisher.java）：

[java] view plain copy

print?

1. /**
2.  * XXX.com Inc.
3.  * Copyright (c) 2004-2015 All Rights Reserved.
4.  */
5. package com.test.SpringTest.activemqinaction.ch4;  
7. import java.util.Hashtable;  
8. import java.util.Map;  
10. import javax.jms.Connection;  
11. import javax.jms.ConnectionFactory;  
12. import javax.jms.Destination;  
13. import javax.jms.JMSException;  
14. import javax.jms.MapMessage;  
15. import javax.jms.Message;  
16. import javax.jms.MessageProducer;  
17. import javax.jms.Session;  
19. import org.apache.activemq.ActiveMQConnectionFactory;  
20. import org.apache.activemq.command.ActiveMQMapMessage;  
22. /**
23.  * 消息产生者
24.  * 
25.  * @author jiangnan
26.  * @version $Id: Publisher.java, v 0.1 2015年7月4日 下午4:48:48 jiangnan Exp $
27.  */
28. public class Publisher {  
30. protected int                                MAX_DELTA_PERCENT = 1;  
31. protected Map<String, Double>                LAST_PRICES       = new Hashtable<String, Double>();  
32. protected static int                         count             = 10;  
33. protected static int                         total;  
35. protected static String                      brokerURL         = "tcp://localhost:61616";  
36. protected static transient ConnectionFactory factory;  
37. protected transient Connection               connection;  
38. protected transient Session                  session;  
39. protected transient MessageProducer          producer;  
41. public Publisher() throws JMSException {  
42.         factory = new ActiveMQConnectionFactory(brokerURL);  
43.         connection = factory.createConnection();  
44.         connection.start();  
45.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
46.         producer = session.createProducer(null);  
47.     }  
49. public void close() throws JMSException {  
50. if (connection != null) {  
51.             connection.close();  
52.         }  
53.     }  
55. public static void main(String[] args) throws JMSException {  
56.         String[] topics = { "topic1", "topic2" };  
57.         Publisher publisher = new Publisher();  
58. while (total < 1000) {  
59. for (int i = 0; i < count; i++) {  
60.                 publisher.sendMessage(topics);  
61.             }  
62.             total += count;  
63.             System.out.println("Published '" + count + "' of '" + total + "' price messages");  
64. try {  
65.                 Thread.sleep(1000);  
66.             } catch (InterruptedException x) {  
67.             }  
68.         }  
69.         publisher.close();  
70.     }  
72. protected void sendMessage(String[] stocks) throws JMSException {  
73. int idx = 0;  
74. while (true) {  
75.             idx = (int) Math.round(stocks.length * Math.random());  
76. if (idx < stocks.length) {  
77. break;  
78.             }  
79.         }  
80.         String stock = stocks[idx];  
81.         Destination destination = session.createTopic("STOCKS." + stock);  
82.         Message message = createStockMessage(stock, session);  
83.         System.out.println("Sending: " + ((ActiveMQMapMessage) message).getContentMap()  
84.                            + " on destination: " + destination);  
85.         producer.send(destination, message);  
86.     }  
88. protected Message createStockMessage(String stock, Session session) throws JMSException {  
89.         Double value = LAST_PRICES.get(stock);  
90. if (value == null) {  
91.             value = new Double(Math.random() * 100);  
92.         }  
94. // lets mutate the value by some percentage
95. double oldPrice = value.doubleValue();  
96.         value = new Double(mutatePrice(oldPrice));  
97.         LAST_PRICES.put(stock, value);  
98. double price = value.doubleValue();  
100. double offer = price * 1.001;  
102. boolean up = (price > oldPrice);  
104.         MapMessage message = session.createMapMessage();  
105.         message.setStringProperty("stock", stock);//设置消息的属性
106.         message.setString("stock", stock);  
107.         message.setDouble("price", price);  
108.         message.setDouble("offer", offer);  
109.         message.setBoolean("up", up);  
110. return message;  
111.     }  
113. protected double mutatePrice(double price) {  
114. double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;  
116. return price * (100 + percentChange) / 100;  
117.     }  
119. }  

消息订阅者（Consumer.java）:

[java] view plain copy

print?

1. /**
2.  * XXX.com Inc.
3.  * Copyright (c) 2004-2015 All Rights Reserved.
4.  */
5. package com.test.SpringTest.activemqinaction.ch4;  
7. import javax.jms.Connection;  
8. import javax.jms.ConnectionFactory;  
9. import javax.jms.Destination;  
10. import javax.jms.JMSException;  
11. import javax.jms.MessageConsumer;  
12. import javax.jms.Session;  
14. import org.apache.activemq.ActiveMQConnectionFactory;  
16. /**
17.  * 消息消费者
18.  * 
19.  * @author jiangnan
20.  * @version $Id: Consumer.java, v 0.1 2015年7月4日 下午4:37:48 jiangnan Exp $
21.  */
22. public class Consumer {  
24. private static String                      brokerURL = "tcp://localhost:61617";  
25. private static transient ConnectionFactory factory;  
26. private transient Connection               connection;  
27. private transient Session                  session;  
29. public Consumer() throws JMSException {  
30.         factory = new ActiveMQConnectionFactory(brokerURL);  
31.         connection = factory.createConnection();  
32.         connection.start();  
33.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
34.     }  
36. public void close() throws JMSException {  
37. if (connection != null) {  
38.             connection.close();  
39.         }  
40.     }  
42. public static void main(String[] args) throws JMSException {  
43.         String[] topics = { "topic1", "topic2" };  
44.         Consumer consumer = new Consumer();  
45. for (String stock : topics) {  
46.             Destination destination = consumer.getSession().createTopic("STOCKS." + stock);  
47. //只接收部分消息的选择器
48.             String selector = "stock = 'topic1'";  
49.             MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination,  
50.                 selector);  
51.             messageConsumer.setMessageListener(new Listener());  
52.         }  
53.     }  
55. public Session getSession() {  
56. return session;  
57.     }  
59. }  

消息监听器（Listener.java）：

[java] view plain copy

print?

1. /**
2.  * XXX.com Inc.
3.  * Copyright (c) 2004-2015 All Rights Reserved.
4.  */
5. package com.test.SpringTest.activemqinaction.ch4;  
7. import java.text.DecimalFormat;  
9. import javax.jms.MapMessage;  
10. import javax.jms.Message;  
11. import javax.jms.MessageListener;  
13. /**
14.  * 消息监听器
15.  * 
16.  * @author jiangnan
17.  * @version $Id: Listener.java, v 0.1 2015年7月4日 下午4:40:25 jiangnan Exp $
18.  */
19. public class Listener implements MessageListener {  
21. public void onMessage(Message message) {  
22. try {  
23.             MapMessage map = (MapMessage) message;  
24.             String stock = map.getString("stock");  
25. double price = map.getDouble("price");  
26. double offer = map.getDouble("offer");  
27. boolean up = map.getBoolean("up");  
28.             DecimalFormat df = new DecimalFormat("#,###,###,##0.00");  
29.             System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t"
30.                                + (up ? "up" : "down"));  
31.         } catch (Exception e) {  
32.             e.printStackTrace();  
33.         }  
34.     }  
36. }  

运行过程： 先启动brokerB： D:\apache-activemq-5.8.0\bin>activemq xbean:file:D:/code/test/SpringTest/src/main/resource/META-INF/spring/activemqinaction/ch4/brokerB.xml

再启动brokerA： D:\apache-activemq-5.8.0\bin>activemq xbean:file:D:/code/test/SpringTest/src/main/resource/META-INF/spring/activemqinaction/ch4/brokerA.xml

可以看到brokerA和brokerB建立连接成功。

然后启动消费者和生产者。在控制台可以观察到消息发送和接收的日志。

Static protocol的使用场景

考虑这样一种场景，多个远程客户端与本地的一个broker建立连接。考虑到每个远程区域的客户端数量，与本地broker关联的连接数会很多。这会给网络带来不必要的负担。为了减小连接数，可以在每个远程区域设置一个broker，然后在远程broker和本地broker之间建立静态链接。这不仅会减小网络连接数，也会提高客户端工作效率。同时也会减少延时，降低等待客户端的时间。

Failover connector

概念介绍

在之前的例子中，客户端仅仅连接到一个特定的broker。如果连接失败或中断，怎么办？有两个选择：客户端会消亡，或者是重新连接到这个broker或者其他broker然后恢复工作。failover可以实现自动重连。有两种方式可以为客户端提供可以连接的broker，一是提供一个静态列表，二是使用动态发现机制。

静态列表配置格式如下： failover:(uri1,...,uriN)?key=value 或者 failover:uri1,...,uriN 默认情况下，会随机挑选可使用的connector。如果连接失败，会挑选另一个URI尝试建立连接。默认配置实现了重连延迟逻辑：第一次重试失败后延迟10ms，之后延迟时间都在前一次的时间之上加倍，直至30000ms。 使用场景 强烈推荐为所有客户端使用failover，即时客户端只会连接到一个broker。这样做的好处是，broker挂掉之后不用手动重新连接，broker恢复后客户端会自动进行重连。简单的利用ActiveMQ的这一特性可以使应用更加稳定。

参考资料：

《ActiveMQ in Action》

附:demo的pom.xml配置

[html] view plain copy

print?

1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
3. <modelVersion>4.0.0</modelVersion>
5. <groupId>com.test</groupId>
6. <artifactId>SpringTest</artifactId>
7. <version>0.0.1-SNAPSHOT</version>
8. <packaging>jar</packaging>
10. <name>SpringTest</name>
11. <url>http://maven.apache.org</url>
13. <properties>
14. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15. <org.springframework.version>3.1.2.RELEASE</org.springframework.version>
16. </properties>
18. <dependencies>
19. <dependency>
20. <groupId>junit</groupId>
21. <artifactId>junit</artifactId>
22. <version>4.11</version>
23. <scope>test</scope>
24. </dependency>
26. <!-- JMX -->
27. <dependency>
28. <groupId>com.sun.jdmk</groupId>
29. <artifactId>jmxtools</artifactId>
30. <version>1.2.1</version>
31. </dependency>
33. <!-- ActiveMQ -->
34. <dependency>
35. <groupId>org.apache.activemq</groupId>
36. <artifactId>activemq-all</artifactId>
37. <version>5.6.0</version>
38. </dependency>
40. <dependency>
41. <groupId>org.apache.activemq</groupId>
42. <artifactId>activemq-core</artifactId>
43. <version>5.5.0</version>
44. </dependency>
46. <!-- xbean (ActiveMQ dependency) -->
47. <dependency>
48. <groupId>org.apache.xbean</groupId>
49. <artifactId>xbean-spring</artifactId>
50. <version>3.7</version>
51. </dependency>
53. <!-- Spring -->
54.     <!--  
55.         Core utilities used by other modules.  
56.         Define this if you use Spring Utility APIs (org.springframework.core.*/org.springframework.util.*)  
57.     -->
58. <dependency>
59. <groupId>org.springframework</groupId>
60. <artifactId>spring-core</artifactId>
61. <version>${org.springframework.version}</version>
62. </dependency>
63.     <!--  
64.         Expression Language (depends on spring-core)  
65.         Define this if you use Spring Expression APIs (org.springframework.expression.*)  
66.     -->
67. <dependency>
68. <groupId>org.springframework</groupId>
69. <artifactId>spring-expression</artifactId>
70. <version>${org.springframework.version}</version>
71. </dependency>
72.     <!--   
73.         Bean Factory and JavaBeans utilities (depends on spring-core)  
74.         Define this if you use Spring Bean APIs (org.springframework.beans.*)   
75.     -->
76. <dependency>
77. <groupId>org.springframework</groupId>
78. <artifactId>spring-beans</artifactId>
79. <version>${org.springframework.version}</version>
80. </dependency>
81.     <!--  
82.         Aspect Oriented Programming (AOP) Framework (depends on spring-core, spring-beans)  
83.         Define this if you use Spring AOP APIs (org.springframework.aop.*)  
84.     -->
85. <dependency>
86. <groupId>org.springframework</groupId>
87. <artifactId>spring-aop</artifactId>
88. <version>${org.springframework.version}</version>
89. </dependency>
90.     <!--  
91.         Application Context (depends on spring-core, spring-expression, spring-aop, spring-beans)   
92.         This is the central artifact for Spring's Dependency Injection Container and is generally always defined  
93.     -->
94. <dependency>
95. <groupId>org.springframework</groupId>
96. <artifactId>spring-context</artifactId>
97. <version>${org.springframework.version}</version>
98. </dependency>
99.     <!--  
100.         Various Application Context utilities, including EhCache, JavaMail, Quartz, and Freemarker integration  
101.         Define this if you need any of these integrations  
102.     -->
103. <dependency>
104. <groupId>org.springframework</groupId>
105. <artifactId>spring-context-support</artifactId>
106. <version>${org.springframework.version}</version>
107. </dependency>
108.     <!--  
109.         Transaction Management Abstraction (depends on spring-core, spring-beans, spring-aop, spring-context)  
110.         Define this if you use Spring Transactions or DAO Exception Hierarchy  
111.         (org.springframework.transaction.*/org.springframework.dao.*)  
112.     -->
113. <dependency>
114. <groupId>org.springframework</groupId>
115. <artifactId>spring-tx</artifactId>
116. <version>${org.springframework.version}</version>
117. </dependency>
118.     <!--  
119.         JDBC Data Access Library (depends on spring-core, spring-beans, spring-context, spring-tx)  
120.         Define this if you use Spring's JdbcTemplate API (org.springframework.jdbc.*)  
121.     -->
122. <dependency>
123. <groupId>org.springframework</groupId>
124. <artifactId>spring-jdbc</artifactId>
125. <version>${org.springframework.version}</version>
126. </dependency>
127.     <!--  
128.         Object-to-Relation-Mapping (ORM) integration with Hibernate, JPA, and iBatis.  
129.         (depends on spring-core, spring-beans, spring-context, spring-tx)  
130.         Define this if you need ORM (org.springframework.orm.*)  
131.     -->
132. <dependency>
133. <groupId>org.springframework</groupId>
134. <artifactId>spring-orm</artifactId>
135. <version>${org.springframework.version}</version>
136. </dependency>
137.     <!--  
138.         Object-to-XML Mapping (OXM) abstraction and integration with JAXB, JiBX, Castor, XStream, and XML Beans.  
139.         (depends on spring-core, spring-beans, spring-context)  
140.         Define this if you need OXM (org.springframework.oxm.*)  
141.     -->
142. <dependency>
143. <groupId>org.springframework</groupId>
144. <artifactId>spring-oxm</artifactId>
145. <version>${org.springframework.version}</version>
146. </dependency>
147.     <!--  
148.         Web application development utilities applicable to both Servlet and Portlet Environments  
149.         (depends on spring-core, spring-beans, spring-context)  
150.         Define this if you use Spring MVC, or wish to use Struts, JSF, or another web framework with Spring (org.springframework.web.*)  
151.     -->
152. <dependency>
153. <groupId>org.springframework</groupId>
154. <artifactId>spring-web</artifactId>
155. <version>${org.springframework.version}</version>
156. </dependency>
157.     <!--  
158.         Spring MVC for Servlet Environments (depends on spring-core, spring-beans, spring-context, spring-web)  
159.         Define this if you use Spring MVC with a Servlet Container such as Apache Tomcat (org.springframework.web.servlet.*)  
160.     -->
161. <dependency>
162. <groupId>org.springframework</groupId>
163. <artifactId>spring-webmvc</artifactId>
164. <version>${org.springframework.version}</version>
165. </dependency>
166.     <!--  
167.         Spring MVC for Portlet Environments (depends on spring-core, spring-beans, spring-context, spring-web)  
168.         Define this if you use Spring MVC with a Portlet Container (org.springframework.web.portlet.*)  
169.     -->
170. <dependency>
171. <groupId>org.springframework</groupId>
172. <artifactId>spring-webmvc-portlet</artifactId>
173. <version>${org.springframework.version}</version>
174. </dependency>
175.     <!--  
176.         Support for testing Spring applications with tools such as JUnit and TestNG  
177.         This artifact is generally always defined with a 'test' scope for the integration testing framework and unit testing stubs  
178.     -->
179. <dependency>
180. <groupId>org.springframework</groupId>
181. <artifactId>spring-test</artifactId>
182. <version>${org.springframework.version}</version>
183. <scope>test</scope>
184. </dependency>
186. <dependency>
187. <groupId>org.springframework</groupId>
188. <artifactId>spring-jms</artifactId>
189. <version>3.1.1.RELEASE</version>
190. </dependency>
192. </dependencies>
193. </project>