activemq--可持久化机制之JDBC

JDBC存储机制

通过数据库的方式，存储消息.

mysql实现

1.添加jar包到activemq安装目录的lib路径下，需要mysql和对应的数据库连接池驱jar包

​ mysql驱动包

​ common-dbcp

​ common-pool

2.修改activemq.xml

将默认的KahaDB持久化机制改为JDBC

 <persistenceAdapter>
     <jdbcPersistenceAdapter dataDirectory="activemq-data" useDatabaseLock="false" dataSource="#mysql-ds" createTablesOnStartup="true" />
</persistenceAdapter>

dataSource指定将要引用的持久化数据库的bean名称，通过#引用

createTablesOnStartup默认是true，表示是否在启动的时候创建数据表，一般在第一次启动的时候设置true，之后改为false

3.添加mysql配置，修改成自己的mysql配置

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
  <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
  <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
  <property name="username" value="activemq"/>
  <property name="password" value="activemq"/>
  <property name="poolPreparedStatements" value="true"/>
</bean>

改配置要添加在<broker>...</broker>标签外面，建议直接在</broker>标签后加上此配置

4.建库建表

在mysql数据库里建立一个activemq数据库，用别的名字也可以.

接着会有三张表是要用到的，ACTIVEMQ_MSGS,ACTIVEMQ_ACKS,ACTIVEMQ_LOCK，这三张表是由配置``createTablesOnStartup`好后，代码集成activemq启动后自动生成的。

ACTIVEMQ_MSGS:消息表，queue和topic都存在里面

字段	说明
ID	自增的主键id
CONTAINER	消息的Destination
MSGID_PROD	消息发送者的主键
MSG_SEQ	发送消息的顺序，MSGID_PROD+MSG_SEQ可以组成JMS的MessageID
EXPIRATION	过期时间，存储的是毫秒数
MSG	消息本体的java序列化对象的二进制数据
PRIORITY	优先级，0-9，数值越大优先级越高

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，其他的只能作为备份等待Master Broker。这张表记录哪个Broker是当前的Master Broker.

自动建表报错的话，复制下面的建表语句手动建表

DROP TABLE IF EXISTS `activemq_acks`;

CREATE TABLE `activemq_acks` (
  `CONTAINER` varchar(250) NOT NULL,
  `SUB_DEST` varchar(250) DEFAULT NULL,
  `CLIENT_ID` varchar(250) NOT NULL,
  `SUB_NAME` varchar(250) NOT NULL,
  `SELECTOR` varchar(250) DEFAULT NULL,
  `LAST_ACKED_ID` bigint(20) DEFAULT NULL,
  `PRIORITY` bigint(20) NOT NULL DEFAULT '5',
  `XID` varchar(250) DEFAULT NULL,
  PRIMARY KEY (`CONTAINER`,`CLIENT_ID`,`SUB_NAME`,`PRIORITY`),
  KEY `ACTIVEMQ_ACKS_XIDX` (`XID`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC;


DROP TABLE IF EXISTS `activemq_lock`;

CREATE TABLE `activemq_lock` (
  `ID` bigint(20) NOT NULL,
  `TIME` bigint(20) DEFAULT NULL,
  `BROKER_NAME` varchar(250) DEFAULT NULL,
  PRIMARY KEY (`ID`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

DROP TABLE IF EXISTS `activemq_msgs`;

CREATE TABLE `activemq_msgs` (
  `ID` bigint(20) NOT NULL,
  `CONTAINER` varchar(250) DEFAULT NULL,
  `MSGID_PROD` varchar(250) DEFAULT NULL,
  `MSGID_SEQ` bigint(20) DEFAULT NULL,
  `EXPIRATION` bigint(20) DEFAULT NULL,
  `MSG` longblob,
  `PRIORITY` bigint(20) DEFAULT NULL,
  `XID` varchar(250) DEFAULT NULL,
  PRIMARY KEY (`ID`),
  KEY `ACTIVEMQ_MSGS_EIDX` (`EXPIRATION`),
  KEY `ACTIVEMQ_MSGS_PIDX` (`PRIORITY`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;