1、将mysql的驱动包复制到activemq的lib目录下
将mysql驱动包（驱动名称：mysql-connector-java-5.1.46.jar）复制到activemq的lib文件夹内
2、配置persistenceAdapter的jdbc
进入到安装目录根目录下conf文件夹中，打开activemq.xml，修改成jdbc的持久化
代码语言：javascript复制<persistenceAdapter>
<!--dataSource将要引用的数据库bean名称，#相当于spring配置文件中的ref=""-->
<!--createTablesOnStartup是否在启动的时候创建数据库表，默认是true-->
<!--true为每次启动都会创建表，一般第一次启动时设置为true，之后改为false-->
<jdbcPersistenceAdapter dataSource="#my-ds" createTablesOnStartup="true"/>
</persistenceAdapter>修改之前：
修改后：
3、数据库连接池配置
将下面配置到activemq.xml文件中，注意需要放在</broker>标签和<import>标签中间
代码语言：javascript复制 <bean id="my-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://192.168.0.102:3306/activemq?relaxAutoCommit=true"/>
<property name="username" value="root"/>
<property name="password" value="root"/>
<property name="poolPreparedStatements" value="true"/>
</bean> 修改前
修改后
4、创建数据库以及表结构
1、创建一个数据库名字为：activemq
2、如果创建好数据库，配置完成，运行mq会自动生成三张表
3、三张表分别为
ACTIVEMQ_MSGS
ACTIVEMQ_ACKS
ACTIVEMQ_LOCK
5、代码运行验证
队列
当消息为非持久化，消息会被保存在内存当中，当消息为持久化时，消息会被保存在文件或者数据库当中，消息一旦被消费就会被删除
生产者
代码语言：javascript复制import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/28 15:58
* @Version: 1.0
*/
public class QueueProduce {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//队列名称
private static final String QUEUE_NAME="queue01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、创建目的地（具体是队列还是主题）,这里是创建队列
Queue queue=session.createQueue(QUEUE_NAME);
//5、创建消息生产者
MessageProducer messageProducer = session.createProducer(queue);
//设置消息持久化
messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
//6、通过messageProducer生产三条消息发送到MQ消息队列中
for (int i=0;i<3;i ){
//7、创建消息
TextMessage textMessage = session.createTextMessage("msg----->" i);//创建一个文本消息
//8、通过messageProducer发送给mq
messageProducer.send(textMessage);
}
messageProducer.close();
session.close();
connection.close();
System.out.println("消息发送成功");
} catch (JMSException e) {
e.printStackTrace();
}
}
}消费者
代码语言：javascript复制import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/28 15:59
* @Version: 1.0
*/
public class QueueConsumer {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//队列名称
private static final String QUEUE_NAME="queue01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、这里接受的queue的名称要和发送者的一致
Queue queue = session.createQueue(QUEUE_NAME);
//5、创建消费者
MessageConsumer consumer = session.createConsumer(queue);
//6、获取消息
while(true){
//MessageConsumer 调用的receive方法为同步调用，在消息到达之前一直阻塞线程
//用什么格式发送，这里就用什么格式接受
//receive等待消息，不限制时间
TextMessage message=(TextMessage)consumer.receive();
//receive带参数等待消息，限制时间，单位毫秒
//TextMessage message=(TextMessage)consumer.receive(4000L);
if(null != message){
System.out.println("接受的消息为------>" message.getText());
}else{
break;
}
}
//7、闭资源
consumer.close();
session.close();
connection.close();
}catch (Exception e){
e.printStackTrace();
}
}
}主题
生产者
代码语言：javascript复制import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/28 16:34
* @Version: 1.0
*/
public class TopicProduce {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//主题名称
private static final String TOPIC_NAME = "topic01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、创建目的地（具体是队列还是主题）,这里是创建主题
Topic topic=session.createTopic(TOPIC_NAME);
//5、创建消息生产者，主题模式
MessageProducer messageProducer = session.createProducer(topic);
//持久化数据
messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
connection.start();
//6、通过messageProducer生产三条消息发送到MQ消息主题中
for (int i=0;i<3;i ){
//7、创建消息
TextMessage textMessage = session.createTextMessage("msg----->" i);//创建一个文本消息
//8、通过messageProducer发送给mq
messageProducer.send(textMessage);
}
messageProducer.close();
session.close();
connection.close();
System.out.println("消息发送成功");
} catch (JMSException e) {
e.printStackTrace();
}
}
}消费者
代码语言：javascript复制import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/28 16:34
* @Version: 1.0
*/
public class TopicConsumer {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//主题名称
private static final String TOPIC_NAME = "topic01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
//订阅名称
connection.setClientID("test01");
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、这里接受的queue的名称要和发送者的一致
Topic topic = session.createTopic(TOPIC_NAME);
//5、持久化的订阅者，第一个参数为订阅主题名称，第二个为备注
TopicSubscriber topicSubscriber=session.createDurableSubscriber(topic,"remakr");
//6、启动连接
connection.start();
//receive等待消息，不限制时间
Message message = topicSubscriber.receive();
while (null!=message){
TextMessage textMessage=(TextMessage)message;
System.out.println("******------>" textMessage.getText());
//等待五秒钟
message=topicSubscriber.receive(5000L);
}
session.close();
connection.close();
}catch (Exception e){
e.printStackTrace();
}
}
}6、查看数据库
队列发送消息数据库中的数据如下图
消费消息之前
消费消息之后
主题消费者启动后数据库如下图
主题发送消息，即使消息被消费，数据库中也会进行保存
7、总结
1、队列消息消费完之后，数据库当中会删除，主题消息消费完之后，消息不会删除
2、 ACTIVEMQ_MSGS保存的队列和消费者消息
3、 ACTIVEMQ_ACKS保存的订阅者
8、注意事项
1、记得需要将用到的相关jar包放到ActiveMQ安装目录下的lib文件夹中，mysql-jdbc的驱动以及对应的连接池jar包，默认会有dbcp的连接池jar
2、createTablesOnStartup属性：在<jdbcPersistenceAdapter> 标签中设置createTablesOnStartup属性为true时在第一次启动ActiveMQ时，会在数据库当中创建所需要的三个表结构，启动完成后去掉或者将该属性改为false
9、JDBC Message store with ActiveMQ journal
1、配置，先将上面的jdbc配置文件注释掉或者删掉，将下面代码复制进去
代码语言：javascript复制<persistenceFactory>
<journalPersistenceAdapterFactory
journalLogFiles="4"
journalLogFileSize="32768"
useJournal="true"
useQuickJournal="true"
dataSource="#my-ds"
dataDirectory="activemq-data"/>
</persistenceFactory>修改前
修改后
2、在队列中刚生产后的消息不会立马同步到数据库
