当前位置：首页 > news >正文

ActiveMQ 源码剖析：消息存储与通信协议实现（二）

news 来源：原创 2025/8/17 23:02:13

四、KahaDB 消息存储实现细节

（一）存储原理分析

KahaDB 作为 ActiveMQ 从 5.4 版本开始的默认消息存储引擎，其基于日志文件的存储原理具有独特的设计和优势。在 KahaDB 的存储目录（如${activemq.data}/kahadb）下，主要包含以下关键文件，它们共同构成了 KahaDB 的存储体系：

db-*.log：这是数据日志文件，用于按顺序存储消息内容。消息以追加的方式写入这些日志文件，充分利用了磁盘的顺序 I/O 特性，极大地提高了写入性能。每个日志文件有默认的大小限制，通常为 32MB ，当一个日志文件写满时，会创建新的日志文件继续写入，文件名中的数字会递增，如db-1.log、db-2.log等。这种设计使得消息写入高效且有序，即使在高并发的消息写入场景下，也能保证良好的性能。例如，在一个电商订单处理系统中，大量的订单消息可以快速地顺序写入db-*.log文件，确保消息的及时存储。

db.data：该文件以 B 树结构存储消息索引。B 树索引通过消息 ID 建立对db-*.log中消息的引用，能够快速定位到消息在日志文件中的位置，从而实现高效的消息检索。当消费者需要获取某条特定 ID 的消息时，系统可以通过db.data中的 B 树索引快速找到该消息在db-*.log中的存储位置，大大提高了消息读取的速度。例如，在一个分布式系统中，多个服务之间通过 ActiveMQ 进行消息通信，当某个服务需要获取特定消息时，借助db.data的索引功能，可以迅速定位到所需消息。

db.redo：用于在 KahaDB 消息存储强制退出后启动时恢复 B 树索引。当系统发生异常（如突然断电、服务器崩溃等）导致 KahaDB 非正常关闭时，db.redo文件中的信息可以帮助系统重建 B 树索引，确保数据的完整性和可恢复性。这就好比一个数据备份，在关键时刻能够恢复系统的关键数据结构，保证消息存储的可靠性。

lock：在集群环境下，用于表示当前获得 KahaDB 读写权限的 broker 。只有获得锁的 broker 才有权限对 KahaDB 进行读写操作，这样可以避免多个 broker 同时对 KahaDB 进行读写操作导致的数据不一致问题。例如，在一个多节点的 ActiveMQ 集群中，通过lock文件可以协调各个节点对 KahaDB 的访问，保证数据的一致性和正确性。

与 AMQ 存储相比，KahaDB 在文件组织和索引结构上有显著差异。AMQ 存储为每个索引使用两个分隔文件，且每个目的地都有一个索引，这在代理拥有大量队列时，会导致文件描述符的大量使用和管理复杂性增加。而 KahaDB 使用单个db.data文件存储所有目的地的 B 树索引，减少了文件描述符的使用，提高了存储的紧凑性和管理的便利性。在消息持久化流程上，KahaDB 通过事务日志和索引文件的配合，确保消息的持久化和快速检索，相比 AMQ 存储，具有更快的存储恢复机制，能够更好地应对系统异常情况，保证消息的可靠性和完整性。

（二）存储过程中的数据操作

消息写入：当生产者发送消息时，KahaDB 首先将消息追加写入db-*.log文件。消息以顺序写入的方式，充分利用磁盘顺序 I/O 的高性能优势。同时，在db.data文件中创建对应的 B 树索引，将消息 ID 与消息在db-*.log中的位置进行关联。例如，在一个实时监控系统中，传感器不断产生大量的监控数据消息，这些消息会快速地顺序写入db-*.log文件，同时在db.data中建立索引，以便后续快速查询和处理。为了保证数据的一致性，KahaDB 在写入过程中会根据enableJournalDiskSyncs配置项（默认为 true）决定是否将消息同步写入磁盘。如果设置为 true，消息会在写入磁盘后才返回确认给生产者，确保消息在磁盘上的持久化；如果设置为 false，写入性能会有所提升，但在系统崩溃等异常情况下，可能会导致部分未同步的消息丢失。

消息读取：消费者读取消息时，首先通过db.data中的 B 树索引根据消息 ID 定位到消息在db-*.log中的位置。然后从db-*.log文件中读取相应的消息内容。在这个过程中，KahaDB 会利用缓存机制（如indexCacheSize配置的索引缓存）来提高读取性能。如果缓存中已经存在所需的索引信息，就可以直接从缓存中获取，避免了磁盘 I/O 操作。例如，在一个电商订单查询系统中，用户查询订单相关消息时，系统可以通过索引快速从db-*.log中读取到对应的消息，同时利用缓存加速索引查找，提高查询效率。如果消息是持久订阅的，KahaDB 还会根据activemq_acks表（如果使用 JDBC 存储时）或相关的订阅信息存储（如在 KahaDB 自身存储机制中）来管理订阅关系和消息的消费进度。

消息删除：当消息被成功消费后，KahaDB 会将其从存储中删除。对于db-*.log文件中的消息，当该文件中的所有消息都被消费后，文件会被标记为可删除，在后续的清理阶段，该文件会被删除或归档，以释放磁盘空间。在删除消息时，KahaDB 会同时更新db.data中的 B 树索引，确保索引的一致性。例如，在一个任务处理系统中，任务完成后对应的消息被消费并删除，此时 KahaDB 会及时清理相关的存储和索引，保证系统的存储资源得到合理利用。如果消息是事务性的，KahaDB 会根据事务的状态进行相应的处理，确保事务的原子性和数据的一致性。若事务提交成功，才会真正删除消息；若事务回滚，则消息会保留在存储中。

五、LevelDB 消息存储实现细节

（一）独特特性分析

LevelDB 是从 ActiveMQ 5.8 之后引入的一种基于文件的本地数据库存储形式，它的出现为 ActiveMQ 的消息存储带来了新的性能提升和特性优化。LevelDB 之所以被引入，主要是为了满足对消息存储更高性能的需求。它基于 Google 开源的 LevelDB 库，针对消息存储场景进行了优化，提供了比 KahaDB 更快的持久性。

LevelDB 采用了独特的存储结构和算法。它将索引保存到包含消息的日志文件中，并且使用了基于跳跃表（Skip List）的数据结构和对磁盘上的数据日志结构进行归并（LSM）的核心操作策略。当 LevelDB 接收到新的消息时，会同步写两个地方：内存中的 MemTable 区域和磁盘上的 Log 文件。直接写 Log 文件是为了在系统异常退出并重启时，能够将 LevelDB 恢复到退出前的结构；同时将消息写入内存的 MemTable 区域，MemTable 区域的数据组织结构是跳跃表，这样可以在读取内存中信息时快速完成信息定位。当 MemTable 区域的数据达到一定的阈值时，会将其转换为 Immutable MemTable，并生成一个新的 MemTable 用于接收新的写入操作，Immutable MemTable 会被异步地刷写到磁盘上，形成一个新的 SSTable（Sorted String Table）文件。

在实际应用中，LevelDB 的高性能优势得到了充分体现。在一个对消息处理实时性要求极高的金融交易系统中，使用 LevelDB 作为消息存储，能够快速地持久化大量的交易消息，确保消息不丢失的同时，还能快速响应消费者对消息的读取请求，大大提高了系统的交易处理效率和稳定性。与 KahaDB 相比，LevelDB 在写入性能上有显著提升，尤其在高并发写入场景下，能够减少磁盘 I/O 操作，提高系统的吞吐量。而且，LevelDB 在垃圾回收机制上也更为高效，能够减少垃圾回收过程中对系统性能的影响。

（二）与其他存储方式的对比

与 KahaDB 对比：LevelDB 和 KahaDB 都是基于文件的存储方式，但它们在实现细节和性能表现上有明显差异。KahaDB 使用自定义的 B 树实现来索引写前日志，而 LevelDB 使用基于 LevelDB 的索引。在性能方面，LevelDB 的写入性能通常优于 KahaDB，因为它采用的 LSM 树结构和跳跃表索引，使得写入操作能够更高效地进行。在高并发写入场景下，LevelDB 能够更快地将消息持久化到磁盘，减少了写入延迟。但在读取性能上，两者各有优劣，KahaDB 的 B 树索引在某些查询场景下可能更适合快速定位消息。从适用场景来看，KahaDB 适用于大多数常规的消息存储场景，它的稳定性和成熟度较高；而 LevelDB 更适合对写入性能要求极高，且对数据一致性和读写性能有特定需求的场景，如实时数据处理系统、高频交易系统等。

与 AMQ 对比：AMQ 是 ActiveMQ 早期的存储方式，与 LevelDB 相比，它在文件组织和索引方式上有很大不同。AMQ 为每个索引使用两个分隔文件，且每个目的地都有一个索引，这在代理拥有大量队列时，会导致文件描述符的大量使用和管理复杂性增加。而 LevelDB 的存储结构更为紧凑和高效。在性能上，AMQ 在写入速度上有一定优势，但在索引重建和文件管理方面存在不足。当 Broker 崩溃时，AMQ 重建索引的速度非常慢，而 LevelDB 由于其独特的存储结构和恢复机制，能够更快地恢复数据。适用场景上，AMQ 适用于早期版本的 ActiveMQ，以及对性能要求不是特别高、数据量较小的简单场景；而 LevelDB 则更适合现代高性能、大规模的消息存储需求。

与 JDBC 对比：JDBC 存储将消息持久化到数据库中，与 LevelDB 基于文件的存储方式有本质区别。JDBC 存储的优点是可以利用成熟的数据库管理系统的特性，如事务处理、数据备份与恢复等。在一些对数据一致性和事务处理要求严格的企业级应用中，JDBC 存储是一个可靠的选择。在银行转账系统中，需要确保消息的原子性和一致性，JDBC 存储可以借助数据库的事务功能来满足这一需求。但 JDBC 存储的缺点也很明显，数据库的 I/O 操作可能会成为性能瓶颈，尤其是在高并发场景下，数据库的连接和读写操作会带来较大的延迟。而 LevelDB 作为基于文件的本地存储，在性能上通常优于 JDBC 存储，能够更快地进行消息的读写操作。适用场景上，JDBC 存储适合对数据一致性和事务处理要求高，且已经有成熟数据库基础设施的企业级应用；LevelDB 则更适合对性能要求高，对数据一致性要求相对较低的高性能消息处理场景。

六、JDBC 消息存储实现细节

（一）配置与使用

使用 JDBC 进行消息存储时，配置步骤如下：

添加数据库驱动包：根据所使用的数据库类型，将对应的数据库驱动包添加到 ActiveMQ 的lib目录下。如果使用 MySQL 数据库，需要将mysql-connector-java-x.x.x.jar添加到该目录。这一步是为了让 ActiveMQ 能够与数据库建立连接，因为驱动包中包含了与数据库通信所需的类和方法。

修改 activemq.xml 配置文件：在activemq.xml文件中，注释掉原有的其他存储方式（如 KahaDB）的配置，并添加 JDBC 存储的配置。找到<persistenceAdapter>标签，将其内容修改为<jdbcPersistenceAdapter dataSource="#mysql-ds" createTableOnStartup="true"/> 。其中，dataSource属性指定了要引用的数据库连接池的 bean 名称，createTableOnStartup属性表示是否在启动时创建数据表，默认值为true，建议在第一次启动时设置为true，后续启动时改为false，以避免重复创建表的操作。

配置数据库连接池：在activemq.xml文件的</broker>与<import resource="jetty.xml"/>中间添加数据库连接池的配置。如果使用org.apache.commons.dbcp2.BasicDataSource连接池，配置示例如下：

<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">

<property name="driverClassName" value="com.mysql.jdbc.Driver"/>

<property name="url" value="jdbc:mysql://localhost:3306/activemq?relaxAutoCommit=true"/>

<property name="username" value="root"/>

<property name="password" value="123456"/>

<property name="poolPreparedStatements" value="true"/>

</bean>

在这个配置中，driverClassName指定了数据库驱动类，url指定了数据库的连接地址，username和password分别是数据库的用户名和密码，poolPreparedStatements表示是否缓存预编译语句，设置为true可以提高性能。

完成上述配置后，启动 ActiveMQ，它会根据配置连接到指定的数据库，并使用 JDBC 进行消息存储。在数据库中，会自动创建三个表：activemq_msgs用于存储 queue 和 topic 的消息；activemq_acks用于存储持久订阅的信息和最后一个持久订阅接收的消息 ID；activemq_lock在集群环境下才有用，确保只有一个 Broker 可以访问，称为 Master Broker 。

（二）性能考量与优化

性能特点分析：JDBC 存储的性能特点与数据库的性能密切相关。由于消息的读写操作都需要通过数据库的 I/O 来完成，在高并发场景下，数据库的 I/O 操作可能会成为性能瓶颈。频繁的消息写入和读取会导致数据库的负载增加，从而影响消息处理的速度。数据库的事务处理也会消耗一定的资源，尤其是在处理大量事务性消息时，会对性能产生较大影响。不过，JDBC 存储也有其优势，它可以利用成熟的数据库管理系统的特性，如数据备份与恢复、数据一致性保障等。

性能优化建议：

- 合理设置数据库连接池参数：连接池的参数设置对性能有重要影响。maxActive参数设置连接池的最大活动连接数，应根据系统的并发访问量合理设置。如果设置过小，在高并发时可能会导致连接不足，影响消息处理速度；如果设置过大，可能会占用过多的系统资源。maxIdle参数设置连接池的最大空闲连接数，合理设置可以避免过多的空闲连接占用资源。minIdle参数设置连接池的最小空闲连接数，确保在系统低负载时也有一定数量的空闲连接可用。

- 优化 SQL 语句：对用于消息存储和查询的 SQL 语句进行优化。避免使用复杂的查询语句，尽量使用简单高效的 SQL 操作。在查询消息时，使用索引可以大大提高查询效率。可以在activemq_msgs表的常用查询字段（如消息 ID、时间戳等）上创建索引。同时，减少不必要的字段查询，只查询需要的字段，以减少数据传输量。

- 批量操作：在进行消息写入和删除等操作时，尽量采用批量操作。JDBC 提供了批量执行 SQL 语句的方法，通过批量操作可以减少数据库连接的开销，提高操作效率。在写入大量消息时，可以将多条消息的插入操作合并为一个批量插入操作。

- 数据库缓存优化：合理利用数据库的缓存机制。可以调整数据库的缓存参数，如 InnoDB 存储引擎的innodb_buffer_pool_size参数，增加缓存大小，以提高数据读取的命中率，减少磁盘 I/O 操作。还可以使用数据库的查询缓存功能，对一些频繁查询且数据变动较小的查询结果进行缓存。

- 定期清理数据：定期清理数据库中已过期或不再需要的消息数据。可以通过定时任务执行删除操作，避免数据量过大导致数据库性能下降。同时，对数据库进行定期的优化和维护，如整理索引、清理碎片等。