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>

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。