ActiveMQ学习之jdbc消息持久化

ActiveMQ学习之jdbc消息持久化

2、配置persistenceAdapter的jdbc

                进入到安装目录根目录下conf文件夹中,打开activemq.xml,修改成jdbc的持久化

<persistenceAdapter>
  <!--dataSource将要引用的数据库bean名称,#相当于spring配置文件中的ref=""-->
  <!--createTablesOnStartup是否在启动的时候创建数据库表,默认是true-->
  <!--true为每次启动都会创建表,一般第一次启动时设置为true,之后改为false-->
  <jdbcPersistenceAdapter dataSource="#my-ds" createTablesOnStartup="true"/> 
</persistenceAdapter>

                修改之前:

                

                修改后:

                

3、数据库连接池配置

           将下面配置到activemq.xml文件中,注意需要放在</broker>标签和<import>标签中间

 <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、代码运行验证

            队列

                    当消息为非持久化,消息会被保存在内存当中,当消息为持久化时,消息会被保存在文件或者数据库当中,消息一旦被消费就会被删除

                    生产者


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();
        }

    }

}

                    消费者


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();
        }
    }

}

            主题

                        生产者


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();
        }

    }
}

                        消费者


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配置文件注释掉或者删掉,将下面代码复制进去

<persistenceFactory>
    <journalPersistenceAdapterFactory 
        journalLogFiles="4" 
        journalLogFileSize="32768" 
        useJournal="true" 
        useQuickJournal="true"
        dataSource="#my-ds"
        dataDirectory="activemq-data"/>
</persistenceFactory>

                修改前

                

        修改后

                

            2、在队列中刚生产后的消息不会立马同步到数据库