ActiveMQ学习-Network connectors

时间 2019-11-17

标签 activemq 学习 network connectors 栏目 ActiveMQ 繁體版

原文原文链接

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

Network connectors

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

Network connector的XML配置以下：web

<networkConnectors>  
<networkConnector name="default-nc" uri="multicast://default"/>  
</networkConnectors>

一个重要的概念-discovery
discovery：是一个检测远程broker服务的进程。client一般须要感知全部的broker。broker，须要感知其余存在的broker，以创建broker的网络。

当咱们想配置一个broker网络时，首要问题是：咱们知道每一个broker的准确地址吗？若是是，能够以静态的方式配置，将客户端链接到提早定义好的broker URI，这在你想彻底控制全部资源的生产环境中比较常见。

若是客户端以及broker相互不知道彼此的地址，那么必须使用一种discovery机制来发现已有的broker。这种设置在开发环境下比较常见，易于配置和维护。

spring

static network

概念介绍

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

Static connector

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

XML中配置示例：apache

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

程序实例

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

BrokerA与brokerB单向相连，当生产者把消息发送给brokerA时，他们会被投递给有订阅需求的broker。这个时候，会被brokerA投递给brokerB。

详细代码以下。
brokerB配置（brokerB.xml）：网络

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

brokerA配置（brokerA.xml）：session

<beans  
  xmlns="http://www.springframework.org/schema/beans"  
  xmlns:amq="http://activemq.org/config/1.0"  
  xmlns:broker="http://activemq.apache.org/schema/core"  
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd  
  http://activemq.org/config/1.0 http://activemq.apache.org/schema/activemq-core.xsd  
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.2.0.xsd  
  http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">  
  
  <!-- Allows us to use system properties as variables in this configuration file -->  
  <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>  
    
  <broker xmlns="http://activemq.apache.org/schema/core" brokerName="BrokerA" dataDirectory="${activemq.base}/data">  
    
    <!-- The transport connectors ActiveMQ will listen to -->  
    <transportConnectors>  
       <transportConnector name="openwire" uri="tcp://localhost:61616" />  
    </transportConnectors>  
      
    <!-- 定义一个network链接器，链接到其余broker -->  
    <networkConnectors>  
        <networkConnector uri="static:(tcp://localhost:61617)" />  
    </networkConnectors>  
  
  </broker>  
    
</beans>

消息生产者（Publisher.java）：mvc

/** 
 * XXX.com Inc. 
 * Copyright (c) 2004-2015 All Rights Reserved. 
 */  
package com.test.SpringTest.activemqinaction.ch4;  
  
import java.util.Hashtable;  
import java.util.Map;  
  
import javax.jms.Connection;  
import javax.jms.ConnectionFactory;  
import javax.jms.Destination;  
import javax.jms.JMSException;  
import javax.jms.MapMessage;  
import javax.jms.Message;  
import javax.jms.MessageProducer;  
import javax.jms.Session;  
  
import org.apache.activemq.ActiveMQConnectionFactory;  
import org.apache.activemq.command.ActiveMQMapMessage;  
  
/** 
 * 消息产生者 
 *  
 * @author jiangnan 
 * @version $Id: Publisher.java, v 0.1 2015年7月4日 下午4:48:48 jiangnan Exp $ 
 */  
public class Publisher {  
  
    protected int                                MAX_DELTA_PERCENT = 1;  
    protected Map<String, Double>                LAST_PRICES       = new Hashtable<String, Double>();  
    protected static int                         count             = 10;  
    protected static int                         total;  
  
    protected static String                      brokerURL         = "tcp://localhost:61616";  
    protected static transient ConnectionFactory factory;  
    protected transient Connection               connection;  
    protected transient Session                  session;  
    protected transient MessageProducer          producer;  
  
    public Publisher() throws JMSException {  
        factory = new ActiveMQConnectionFactory(brokerURL);  
        connection = factory.createConnection();  
        connection.start();  
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
        producer = session.createProducer(null);  
    }  
  
    public void close() throws JMSException {  
        if (connection != null) {  
            connection.close();  
        }  
    }  
  
    public static void main(String[] args) throws JMSException {  
        String[] topics = { "topic1", "topic2" };  
        Publisher publisher = new Publisher();  
        while (total < 1000) {  
            for (int i = 0; i < count; i++) {  
                publisher.sendMessage(topics);  
            }  
            total += count;  
            System.out.println("Published '" + count + "' of '" + total + "' price messages");  
            try {  
                Thread.sleep(1000);  
            } catch (InterruptedException x) {  
            }  
        }  
        publisher.close();  
    }  
  
    protected void sendMessage(String[] stocks) throws JMSException {  
        int idx = 0;  
        while (true) {  
            idx = (int) Math.round(stocks.length * Math.random());  
            if (idx < stocks.length) {  
                break;  
            }  
        }  
        String stock = stocks[idx];  
        Destination destination = session.createTopic("STOCKS." + stock);  
        Message message = createStockMessage(stock, session);  
        System.out.println("Sending: " + ((ActiveMQMapMessage) message).getContentMap()  
                           + " on destination: " + destination);  
        producer.send(destination, message);  
    }  
  
    protected Message createStockMessage(String stock, Session session) throws JMSException {  
        Double value = LAST_PRICES.get(stock);  
        if (value == null) {  
            value = new Double(Math.random() * 100);  
        }  
  
        // lets mutate the value by some percentage  
        double oldPrice = value.doubleValue();  
        value = new Double(mutatePrice(oldPrice));  
        LAST_PRICES.put(stock, value);  
        double price = value.doubleValue();  
  
        double offer = price * 1.001;  
  
        boolean up = (price > oldPrice);  
  
        MapMessage message = session.createMapMessage();  
        message.setStringProperty("stock", stock);//设置消息的属性  
        message.setString("stock", stock);  
        message.setDouble("price", price);  
        message.setDouble("offer", offer);  
        message.setBoolean("up", up);  
        return message;  
    }  
  
    protected double mutatePrice(double price) {  
        double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;  
  
        return price * (100 + percentChange) / 100;  
    }  
  
}

消息订阅者（Consumer.java）:app

/** 
 * XXX.com Inc. 
 * Copyright (c) 2004-2015 All Rights Reserved. 
 */  
package com.test.SpringTest.activemqinaction.ch4;  
  
import javax.jms.Connection;  
import javax.jms.ConnectionFactory;  
import javax.jms.Destination;  
import javax.jms.JMSException;  
import javax.jms.MessageConsumer;  
import javax.jms.Session;  
  
import org.apache.activemq.ActiveMQConnectionFactory;  
  
/** 
 * 消息消费者 
 *  
 * @author jiangnan 
 * @version $Id: Consumer.java, v 0.1 2015年7月4日 下午4:37:48 jiangnan Exp $ 
 */  
public class Consumer {  
  
    private static String                      brokerURL = "tcp://localhost:61617";  
    private static transient ConnectionFactory factory;  
    private transient Connection               connection;  
    private transient Session                  session;  
  
    public Consumer() throws JMSException {  
        factory = new ActiveMQConnectionFactory(brokerURL);  
        connection = factory.createConnection();  
        connection.start();  
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
    }  
  
    public void close() throws JMSException {  
        if (connection != null) {  
            connection.close();  
        }  
    }  
  
    public static void main(String[] args) throws JMSException {  
        String[] topics = { "topic1", "topic2" };  
        Consumer consumer = new Consumer();  
        for (String stock : topics) {  
            Destination destination = consumer.getSession().createTopic("STOCKS." + stock);  
            //只接收部分消息的选择器  
            String selector = "stock = 'topic1'";  
            MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination,  
                selector);  
            messageConsumer.setMessageListener(new Listener());  
        }  
    }  
  
    public Session getSession() {  
        return session;  
    }  
  
}

消息监听器（Listener.java）：dom

/** 
 * XXX.com Inc. 
 * Copyright (c) 2004-2015 All Rights Reserved. 
 */  
package com.test.SpringTest.activemqinaction.ch4;  
  
import java.text.DecimalFormat;  
  
import javax.jms.MapMessage;  
import javax.jms.Message;  
import javax.jms.MessageListener;  
  
/** 
 * 消息监听器 
 *  
 * @author jiangnan 
 * @version $Id: Listener.java, v 0.1 2015年7月4日 下午4:40:25 jiangnan Exp $ 
 */  
public class Listener implements MessageListener {  
  
    public void onMessage(Message message) {  
        try {  
            MapMessage map = (MapMessage) message;  
            String stock = map.getString("stock");  
            double price = map.getDouble("price");  
            double offer = map.getDouble("offer");  
            boolean up = map.getBoolean("up");  
            DecimalFormat df = new DecimalFormat("#,###,###,##0.00");  
            System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t"  
                               + (up ? "up" : "down"));  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
  
}

运行过程：
先启动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配置

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
  <modelVersion>4.0.0</modelVersion>  
  
  <groupId>com.test</groupId>  
  <artifactId>SpringTest</artifactId>  
  <version>0.0.1-SNAPSHOT</version>  
  <packaging>jar</packaging>  
  
  <name>SpringTest</name>  
  <url>http://maven.apache.org</url>  
  
  <properties>  
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>  
    <org.springframework.version>3.1.2.RELEASE</org.springframework.version>  
  </properties>  
  
  <dependencies>  
    <dependency>  
      <groupId>junit</groupId>  
      <artifactId>junit</artifactId>  
      <version>4.11</version>  
      <scope>test</scope>  
    </dependency>  
      
    <!-- JMX -->  
    <dependency>  
        <groupId>com.sun.jdmk</groupId>  
        <artifactId>jmxtools</artifactId>  
        <version>1.2.1</version>  
    </dependency>  
      
    <!-- ActiveMQ -->  
    <dependency>  
      <groupId>org.apache.activemq</groupId>  
      <artifactId>activemq-all</artifactId>  
      <version>5.6.0</version>  
    </dependency>  
      
    <dependency>  
      <groupId>org.apache.activemq</groupId>  
      <artifactId>activemq-core</artifactId>  
      <version>5.5.0</version>  
    </dependency>  
      
    <!-- xbean (ActiveMQ dependency) -->  
    <dependency>  
      <groupId>org.apache.xbean</groupId>  
      <artifactId>xbean-spring</artifactId>  
      <version>3.7</version>  
    </dependency>  
      
    <!-- Spring -->  
    <!--  
        Core utilities used by other modules.  
        Define this if you use Spring Utility APIs (org.springframework.core.*/org.springframework.util.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-core</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Expression Language (depends on spring-core)  
        Define this if you use Spring Expression APIs (org.springframework.expression.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-expression</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--   
        Bean Factory and JavaBeans utilities (depends on spring-core)  
        Define this if you use Spring Bean APIs (org.springframework.beans.*)   
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-beans</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Aspect Oriented Programming (AOP) Framework (depends on spring-core, spring-beans)  
        Define this if you use Spring AOP APIs (org.springframework.aop.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-aop</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Application Context (depends on spring-core, spring-expression, spring-aop, spring-beans)   
        This is the central artifact for Spring's Dependency Injection Container and is generally always defined  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-context</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Various Application Context utilities, including EhCache, JavaMail, Quartz, and Freemarker integration  
        Define this if you need any of these integrations  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-context-support</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Transaction Management Abstraction (depends on spring-core, spring-beans, spring-aop, spring-context)  
        Define this if you use Spring Transactions or DAO Exception Hierarchy  
        (org.springframework.transaction.*/org.springframework.dao.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-tx</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        JDBC Data Access Library (depends on spring-core, spring-beans, spring-context, spring-tx)  
        Define this if you use Spring's JdbcTemplate API (org.springframework.jdbc.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-jdbc</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Object-to-Relation-Mapping (ORM) integration with Hibernate, JPA, and iBatis.  
        (depends on spring-core, spring-beans, spring-context, spring-tx)  
        Define this if you need ORM (org.springframework.orm.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-orm</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Object-to-XML Mapping (OXM) abstraction and integration with JAXB, JiBX, Castor, XStream, and XML Beans.  
        (depends on spring-core, spring-beans, spring-context)  
        Define this if you need OXM (org.springframework.oxm.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-oxm</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Web application development utilities applicable to both Servlet and Portlet Environments  
        (depends on spring-core, spring-beans, spring-context)  
        Define this if you use Spring MVC, or wish to use Struts, JSF, or another web framework with Spring (org.springframework.web.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-web</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Spring MVC for Servlet Environments (depends on spring-core, spring-beans, spring-context, spring-web)  
        Define this if you use Spring MVC with a Servlet Container such as Apache Tomcat (org.springframework.web.servlet.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-webmvc</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Spring MVC for Portlet Environments (depends on spring-core, spring-beans, spring-context, spring-web)  
        Define this if you use Spring MVC with a Portlet Container (org.springframework.web.portlet.*)  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-webmvc-portlet</artifactId>  
      <version>${org.springframework.version}</version>  
    </dependency>  
    <!--  
        Support for testing Spring applications with tools such as JUnit and TestNG  
        This artifact is generally always defined with a 'test' scope for the integration testing framework and unit testing stubs  
    -->  
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-test</artifactId>  
      <version>${org.springframework.version}</version>  
      <scope>test</scope>  
    </dependency>  
      
    <dependency>  
      <groupId>org.springframework</groupId>  
      <artifactId>spring-jms</artifactId>  
      <version>3.1.1.RELEASE</version>  
    </dependency>  
      
  </dependencies>  
</project>