ActiveMQ持久化机制

Posted mayuan01

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ActiveMQ持久化机制相关的知识,希望对你有一定的参考价值。

为什么要进行持久化

  为了避免MQ服务器意外宕机导致数据丢失,需要做到重启后没有被消费的数据依然在消息队列中。

ActiveMQ的持久化机制包含:

  1. JDBC:持久化到数据库

  2. AMQ:日志文件

  3. KahaBD:AMQ基础上改进,默认选择

  4. LevelDB:谷歌K/V数据库

  注:ActiveMQ默认是不开启持久化的。

ActiveMQ默认持久化机制

    <persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb"/>
        </persistenceAdapter>

  性能:

    AMQ的性能改与JDBC的持久化机制,由于是在文件中追加写入消息,所以性能比较高。并且创建了消息主键索引和缓存索引机制以提升性能。

  缺点:

    AMQ会为每一个Destination创建一个索引,若创建了大小的消息队列,则磁盘占用会非常大,所以由于索引文件比较大,当Broker崩溃后,重建所以速度比较慢。  

JDBC持久化机制

  步骤一:创建数据库

    技术图片

  步骤二:在程序中开启持久化操作

producer.setDeliveryMode(DeliveryMode.PERSISTENT);

    PERSISTENT:代表开启持久化

    NON_PERSISTENT:代表不开启持久化

  步骤三:配置activemq.xml文件

    文件位置在conf目录下

    3.1 在persistenceAdapter加入如下配置

     <persistenceAdapter>
        <!--第一次创建是createTablesOnStartup属性为true,之后创建完成后将属性该为false,一直为true的话,每次启动的时候都会创建新表 --> <jdbcPersistenceAdapter dataSource="#activemq-db" createTablesOnStartup="true" /> </persistenceAdapter>

    3.2 配置数据源

  <bean id="activemq-db" class="org.apache.commons.dbcp.BasicDataSource"> <property name="driverClassName" value="com.mysql.jdbc.Driver"/> <property name="url" value="jdbc:mysql://127.0.0.1:3306/activemq"/> <property name="username" value="root"/> <property name="password" value="123"/> <property name="maxActive" value="200"/> <property name="poolPreparedStatements" value="true"/> </bean>

    3.3 添加依赖包

      由于JDBC持久话方式需要连接数据库,所以需要在ActiveMQ安装目录的lib目录下添加如下jar包:

      技术图片

  步骤四:重启

    4.1重新启动activeMQ

      重启之后数据库中将会自动创建三张表

      技术图片

      activemq_acks:用于存储订阅关系。如果是持久化Topic,订阅者和服务器的订阅关系在这个表保存。

        主要的数据库字段如下:

          CONTAINER:消息的Destination

          SUB_DEST:如果是使用Static集群,这个字段会有集群其他系统的信息

          CLIENT_ID:每个订阅者都必须有一个唯一的客户端ID用于区分

          SUB_NAME:订阅者名称

          SELECTOR:选择器,可以选择只消费满足条件的消息。条件可以用自定义属性实现,可支持多属性AND和OR操作

          LAST_ACKED_ID:记录最后消费过的消息的ID

      activemq_lock:在集群环境下才有用,只有一个Broker可以获得消息,称为Master Broker

      activemq_msgs:用于存储消息,Queue和Topic都存储在这个表中;

        主要的数据库字段如下:

          CONTAINER:消息的Destination

          MSGID_PROD:消息发送者客户端的主键

          MSG_SEQ:是发送消息的顺序,MSGID+PROD+MSG_SEQ可以组成JMS的MessageID

          EXPIRATION:消息的过期时间

          MSG:消息本体的java序列化对象的二进制数据

          PRIORITY:优先级,从0-9,数值越大优先级越高

    4.2 启动生产者和消费者,测试效果

      启动生产者:

        技术图片

        生产者发送完成,下面可以查看一下队列中的数据

        技术图片

        因为上面做配置将数据保存在数据库中,下面可以查看一下数据库中的数据

        技术图片

      关闭ActiveMQ服务测试数据有没有真正的持久化

        关闭之后页面时无法访问的;

        技术图片

        关闭之后刷新数据库,数据依然存在,这是数据已经持久化了

      启动ActiveMQ服务以及消费者:

        技术图片

         启动消费者时,数据已经成功接收到了

        技术图片

        因为消费者已经接受到消息了,所以队列中的待处理数据就变成了0,并且现在的消费者是1;

        下面可以查看一个数据库中的数据

        技术图片

        由于消费者已经接受消息,队列中数据删除了,所以数据库中的数据也被删除了

      结论:生产者就数据发送到队列中,如果不做持久化操作,ActiveMQ服务器宕机,数据就会清空;如果做了持久化操作,ActiveMQ服务器宕机之后再次启动,没有被消费者消费的消息就会一直在数据库中,等消费者消费完消息后,数据库中的数据才会被删除。

以上是关于ActiveMQ持久化机制的主要内容,如果未能解决你的问题,请参考以下文章

ActiveMQ持久化机制

ActiveMQ开启用户认证及消息持久化到数据库功能

ActiveMq笔记2-消息持久化

ActiveMQ消息持久化

activemq部分消息无法取走

ActiveMQ服务挂掉怎么办