ActiveMQ持久化机制和JMS可靠消息

1、ActiveMQ持久化机制 1.1 JDBC将数据持久化到数据库 1.2 AMQ生成日志文件 1.3 KahaDB:本次磁盘生成数据文件(默认) 1.4 LevelDB:谷歌K/V数据库 1.5 默认不持久化,正常的开发模式当中,是需要持久化的  

  DeliveryMode.NON_PERSISTENT = 1 不开启持久化
  DeliveryMode.PERSISTENT = 2 开启持久化
  

1.6 使用JDBC持久化   步骤一:创建一个数据库

    

ActiveMQ持久化机制和JMS可靠消息

步骤二:配置activemq.xml配置文件 

1.在persistenceAdapter加入如下配置
    <!--createTablesOnStartup 启动是否创建表 第一次为true 后续为false-->
    <jdbcPersistenceAdapter dataSource="#activemq-db" createTablesOnStartup="true" />
  

ActiveMQ持久化机制和JMS可靠消息

  2.配置数据源

  

ActiveMQ持久化机制和JMS可靠消息

  3.将数据库连接Jar放到activemq解压的lib文件夹下

  

ActiveMQ持久化机制和JMS可靠消息

步骤三:重新启动ActiveMQ

  

ActiveMQ持久化机制和JMS可靠消息

  数据表解释:

  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,数值越大优先级越高

步骤四:启动提供者后数据表activemq_msgs便增加一条数据

  

ActiveMQ持久化机制和JMS可靠消息

   

ActiveMQ持久化机制和JMS可靠消息

步骤五:消费者启动后数据表消息被删除

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

  

ActiveMQ持久化机制和JMS可靠消息

2、JMS可靠消息 2.1 带事务的Session   ①生产者必须在生产完数据后手动提交session

  

ActiveMQ持久化机制和JMS可靠消息

package com.zn.p2p;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 点对点提供者
 */
public class P2P_Provider {
    public static void main(String[] args) throws JMSException {
        //步骤一:创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory();
        //步骤二:创建连接
        Connection connection = activeMQConnectionFactory.createConnection();
        //步骤三:启动连接
        connection.start();
        //步骤四:获取会话工厂
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        //步骤五:创建队列
        Queue quque = session.createQueue("MyQueue");
        //创建消息生产者
        MessageProducer producer = session.createProducer(quque);
        //消息持久化  参数1为不做持久化  2为持久化
        producer.setDeliveryMode(2);
        //模拟消息
        TextMessage textMessage = session.createTextMessage("Hello ActiveMQ!");
        //发送消息
        producer.send(textMessage);
        //手动提交消息
        session.commit();
        System.out.println("生产者生成消息完毕!");
        //回收资源
        session.close();
        connection.close();
    }
}

②消费者在消费完数据之后也必须手动提交Session

  

ActiveMQ持久化机制和JMS可靠消息

package com.zn.p2p;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/52764e6b1c067d517ebed5ed139d1a54.html