ActiveMQ之基础讲解篇.docx
ActiveMQ之基础讲解篇文档编号:2020-03-09当前版本：V1.0.0编写者： 海带发布时间:2020-03-09修订记录板本说明作者批准批准日期1.0C海带2020-03-091.11.21.31.41.51.61.71.81.92.02.12.22.32.42.52.62.72.82.93.03.11.概述41.1ActiveMQ是什么41.2整体架构41.2.1相关概念42.P2P53.Publish64.代码流程65.存储方式75.1KahaDB75.2AMQ85.3JDBC85.4内存存储96.部署模式96.1单例模式96.2无共享主从模式96.3共享存储主从模式116.3.1基于数据库116.3.2基于文件系统127.网络连接137.1代理网络137.2网络发现148.实例159.性能调优151. 概述本文档为海带中间件系列之ActiveMQ基础篇，描述了ActiveMQ的基本概念、运行机制及单Broker、多Broker的使用方法。给大家明确一点，ActiveMQ只是一种工具，安装、配置后使用即可，可以简单地集成到Spring等框架中来，就像我们使用MySql数据库一样。1.1 ActiveMQ是什么ActiveMQ是一种MOM，即消息中间件，是用于以分布式应用或系统中的异步、松耦合、可靠、可扩展和安全通信的一类软件。MOM 的总体思想是它作为消息发送器和消息接收器之间的消息中介,这种中介提供了一个全新水平的松耦合。JMS 叫做 Java 消息服务（Java Message Service）,是 Java 平台上有关面向 MOM 的技术规范，旨在通过提供标准的产生、发送、接收和处理消息的 API 简化企业应用的开发，类似于 JDBC 和关系型数据库通信方式的抽象。1.2 整体架构1.2.1 相关概念l Provider：纯 Java 语言编写的 JMS 接口实现（如 ActiveMQ ）l Domains：消息传递方式，包括点对点（P2P）、发布/订阅（Pub/Sub）两种l Connection factory：客户端使用连接工厂来创建与 JMS provider 的连接l Destination：消息被寻址、发送以及接收的对象2. P2PP2P （点对点）消息域使用 queue 作为 Destination，消息可以被同步或异步的发送和接收，每个消息只会给一个 Consumer 传送一次。Consumer 可以使用 MessageConsumer.receive() 同步地接收消息，也可以通过使用MessageConsumer.setMessageListener() 注册一个 MessageListener 实现异步接收。多个 Consumer 可以注册到同一个 queue 上，但一个消息只能被一个 Consumer 所接收，然后由该 Consumer 来确认消息。并且在这种情况下，Provider 对所有注册的 Consumer 以轮询的方式发送消息。说明：l 消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。l 消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。l Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费，其它的则不能消费此消息了。l 当消费者不存在时，消息会一直保存，直到有消费者消费3. PublishPub/Sub（发布/订阅，Publish/Subscribe）消息域使用 topic 作为 Destination，发布者向 topic 发送消息，订阅者注册接收来自 topic 的消息。发送到 topic 的任何消息都将自动传递给所有订阅者。接收方式（同步和异步）与 P2P 域相同。除非显式指定，否则 topic 不会为订阅者保留消息。当然，这可以通过持久化（Durable）订阅来实现消息的保存。这种情况下，当订阅者与 Provider 断开时，Provider 会为它存储消息。当持久化订阅者重新连接时，将会受到所有的断连期间未消费的消息。说明：l 消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息l 和点对点方式不同，发布到topic的消息会被所有订阅者消费l 当生产者发布消息，不管是否有消费者。都不会保存消息4. 代码流程l 获取连接工厂l 使用连接工厂创建连接l 启动连接l 从连接创建会话l 获取 Destinationl 创建 Producer，或n 创建 Producern 创建 messagel 创建 Consumer，或发送或接收message发送或接收 messagen 创建 Consumern 注册消息监听器（可选）l 发送或接收 messagel 关闭资源（connection, session, producer, consumer 等)注意这张表格：5. 存储方式5.1 KahaDBActiveMQ 5.3 版本起的默认存储方式。KahaDB存储是一个基于文件的快速存储消息，设计目标是易于使用且尽可能快。它使用基于文件的消息数据库意味着没有第三方数据库的先决条件。要启用 KahaDB 存储，需要在 activemq.xml 中进行以下配置：<broker brokerName="broker" persistent="true" useShutdownHook="false"> <persistenceAdapter> <kahaDB directory="$activemq.data/kahadb" journalMaxFileLength="16mb"/> </persistenceAdapter></broker>5.2 AMQ与 KahaDB 存储一样，AMQ存储使用户能够快速启动和运行，因为它不依赖于第三方数据库。AMQ 消息存储库是可靠持久性和高性能索引的事务日志组合，当消息吞吐量是应用程序的主要需求时，该存储是最佳选择。但因为它为每个索引使用两个分开的文件，并且每个 Destination 都有一个索引，所以当你打算在代理中使用数千个队列的时候，不应该使用它。<persistenceAdapter> <amqPersistenceAdapter directory="$activemq.data/kahadb" syncOnWrite="true" indexPageSize="16kb" indexMaxBinSize="100" maxFileLength="10mb" /></persistenceAdapter>5.3 JDBC选择关系型数据库，通常的原因是企业已经具备了管理关系型数据的专长，但是它在性能上绝对不优于上述消息存储实现。事实是，许多企业使用关系数据库作为存储，是因为他们更愿意充分利用这些数据库资源。注意要把Mysql的Jar包拷贝到/lib目录下，并修改my.ini中以下两个属性：log-bin=master-binbinlog_format=mixed<beans> <broker brokerName="test-broker" persistent="true" xmlns="http:/activemq.apache.org/schema/core"> <persistenceAdapter> <jdbcPersistenceAdapter dataDirectory="$activemq.data" cleanupPeriod="0" dataSource="#mysql-ds"/> </persistenceAdapter> </broker> <bean id="mysql-ds" class="mons.dbcp2.BasicDataSource" destroy-method="close"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql:/localhost/activemq?relaxAutoCommit=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Hongkong"/> <property name="username" value="root"/> <property name="password" value="root"/> <property name="maxTotal" value="200"/> <property name="poolPreparedStatements" value="true"/> </bean></beans>5.4 内存存储内存消息存储器将所有持久消息保存在内存中。在仅存储有限数量 Message 的情况下，内存消息存储会很有用，因为 Message 通常会被快速消耗。在 activema.xml 中将 broker 元素上的 persistent 属性设置为 false 即可。<broker brokerName="test-broker" persistent="false" xmlns="http:/activemq.apache.org/schema/core"> <transportConnectors> <transportConnector uri="tcp:/localhost:61635"/> </transportConnectors></broker>6. 部署模式6.1 单例模式太简单，不做实例分析。6.2 无共享主从模式这是最简单的 Provider 高可用性的方案，主从节点分别存储 Message。从节点需要配置为连接到主节点，并且需要特殊配置其状态。所有消息命令（消息，确认，订阅，事务等）都从主节点复制到从节点，这种复制发生在主节点对其接收的任何命令生效之前。并且，当主节点收到持久消息，会等待从节点完成消息的处理（通常是持久化到存储），然后再自己完成消息的处理（如持久化到存储）后，再返回对 Producer 的回执。从节点不启动任何传输，也不能接受任何客户端或网络连接，除非主节点失效。当主节点失效后，从节点自动成为主节点，并且开启传输并接受连接。这是，使用 failover 传输的客户端就会连接到该新主节点。Broker 连接配置如下：failover:/(tcp:/masterhost:61616,tcp:/slavehost:61616)?randomize=false但是，这种部署模式有一些限制，l 主节点只会在从节点连接到主节点时复制其活动状态，因此当从节点没有连接上主节点之前，任何主节点处理的 Message 或者消息确认都会在主节点失效后丢失。不过可以通过在主节点设置 waitForSlave 来避免，这样就强制主节点在没有任何一个从节点连接上的情况下接受连接。l 就是主节点只能有一个从节点，并且从节点不允许再有其他从节点。l 把正在运行的单例配置成无共享主从，或者配置新的从节点时，你都要停止当前服务，修改配置后再重启才能生效。在可以接受一些故障停机时间的情况下，可以使用该模式。从节点配置：<services> <masterConnector remoteURI="tcp:/remotehost:62001" userName="Rob" password="Davies"/></services>此外，可以配置 shutdownOnMasterFailure 项，表示主节点失效后安全关闭，保证没有消息丢失，允许管理员维护一个新的从节点。6.3 共享存储主从模式允许多个代理共享存储，但任意时刻只有一个是活动的。这种情况下，当主节点失效时，无需人工干预来维护应用的完整性。另外一个好处就是没有从节点数的限制。有如下两种模式。6.3.1 基于数据库它会获取一个表上的排它锁，以确保没有其他 ActiveMQ 代理可以同时访问数据库。其他未获得锁的代理则处于轮询状态，就会被当做是从节点，不会开启传输也不会接受连接。6.3.2 基于文件系统需要获取分布式共享文件锁，linux 系统下推荐用 GFS2。7. 网络连接7.1 代理网络支持将 ActiveMQ 消息代理链接到不同拓扑，这就是被人们熟知的代理网络。ActiveMQ 网络使用存储和转发的概念，其中消息总是存储在本地代理中，然后通过网络转发到另一个代理。当连接建立后，远程代理将把包含其所有持久和活动消费者目的地的信息传递给本地代理，本地代理根据信息决定远程代理感兴趣的 Message 并将它发送给远程代理。如果希望网络是双向的，您可以使用网络连接器将远程代理配置为指向本地代理，或将网络连接器配置为双工，以便双向发送消息。<networkConnectors> <networkConnector uri="static:/(tcp:/backoffice:61617)" name="bridge" duplex="true" conduitSubscriptions="true" decreaseNetworkConsumerPriority="false"> </networkConnector></networkConnectors>注意，配置的顺序很重要：1. 网络连接需要在消息存储前建立好连接，对应 networkConnectors 元素2. 消息存储需要在传输前配置好，对应 persistenceAdapter 元素3. 消息传输最后配置，对应 transportConnectors 元素7.2 网络发现1. 动态发现使用多播来支持网络动态发现。配置如下：<networkConnectors><networkConnector uri="multicast:/default"/></networkConnectors>其中，multicast:/ 中的默认名称表示该代理所属的组。因此使用此方式时，强烈推荐你使用一个独特的组名，避免你的代理连接到其他不相关代理。2. 静态发现静态发现接受代理 URI 列表，并将尝试按列表中确定的顺序连接到远程代理。<networkConnectors><networkConnector uri="static:(tcp:/remote-master:61617,tcp:/remote-slave:61617)"/></networkConnectors>相关配置如下：l initialReconnectDelay：默认值1000，表示尝试连接前的时延。l maxReconnectDelay：默认值30000，表示连接失败后到重新建立连接之间的时延，仅在 useExponentialBackOff 启用时生效。l useExponentialBackOff：默认值 true，如果启用，表示每次失败后增加重建连接的时延。l backOffMultiplier：默认值2，表示启用 useExponentialBackOff 后每次的时延增量需要注意的是，网络连接将始终尝试建立到远程代理的连接。需要注意的是，网络连接将始终尝试建立到远程代理的连接。3. 多连接场景当网络负载高时，使用多连接很有意义。但是你需要确保不会重复传递消息，这可以通过过滤器来实现。<networkConnectors><networkConnector uri="static:/(tcp:/remotehost:61617)" name="queues_only" duplex="true"<excludedDestinations><topic physicalName=">"/></excludedDestinations></networkConnector><networkConnector uri="static:/(tcp:/remotehost:61617)" name="topics_only" duplex="true"<excludedDestinations><queue physicalName=">"/></excludedDestinations></networkConnector></networkConnectors>8. 实例太简单，不做实例分析。9. 性能调优ActiveMQ需要关注的性能有三个方面：l 消息从磁盘中写入和读取（仅限与持久消息）l 消息的编组和基于收集的传递l 基于多线程的高低文切换
