ActiveMQ的消息持久化---JDBC的实现方式

太过爱你忘了你带给我的痛 2022-09-25 01:13 196阅读 0赞

对于queue的方式做到持久化很简单。每个消息只有一个消费者，只要将消息存入数据库，然后消费者取走信息即可，在这里我们讨论的是topic的持久化方式。

1.在这里我用的是oracle数据库，先将oracle的驱动包放入lib文件夹下，再配置activemq.xml文件中的配置信息

将原来的默认持久化方式改为使用jdbc的方式：

<persistenceAdapter>
    			<jdbcPersistenceAdapter dataDirectory="${activemq.base}/data" dataSource="#oracle-ds" useDatabaseLock="false"/>
    		</persistenceAdapter>

配置oracle-ds的bean，注意须将bean配置在broker节点之外

2.编写生产者与消费者类

生产者

package com.yc.mytest;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class MyProducer {
    	private static final String URL=ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址
    	private static String TOPIC_NAME="MyTopicTest"; //topic名字
    
    	public static void main(String[] args) {
    		ConnectionFactory factory; //连接工厂
    		Connection conn = null; //连接
    		Session session; //会话
    		Destination dest; //消息的目的地
    		MessageProducer producer; //消息生产者
    
    		try {
    			//初始化连接工厂
    			factory=new ActiveMQConnectionFactory(URL); 
    
    			//获取连接
    			conn=factory.createConnection(); 
    
    			//开始连接
    			conn.start();
    
    			//创建Session，此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
    			session=conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
    			//创建topic,producer和consumer将根据TOPIC_NAME来发送/接收对应的消息 
    			dest=session.createTopic(TOPIC_NAME); 
    
    			//创建生产者
    			producer=session.createProducer(dest);
    
    			//设置持久化
    			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
    			//发送消息
    			sendMessage(session, producer);
    		} catch (Exception e) {
    			System.out.println("出错啦!!!");
    			e.printStackTrace();
    		} finally{
    			if(conn!=null){
    				try {
    					//关闭连接
    					conn.close();
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    	private static void sendMessage(Session session,MessageProducer producer) throws JMSException{
    		//初始化一个mq消息
    		TextMessage message = session.createTextMessage() ;//= session.createTextMessage("你好 ActiveMq " + i);
    		//填写消息内容
    		message.setText("你好 ActiveMq");
    		//发送消息
    		producer.send(message);
    		System.out.println("消息发送成功!!!");
    	}
    }

消费者

package com.yc.mytest;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class MyConsumer implements MessageListener {
    	private static final String URL=ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址
    	private static String TOPIC_NAME="MyTopicTest"; //topic名字
    
    	public static void main(String[] args) {
    		ConnectionFactory factory; //连接工厂
    		Connection conn = null; //连接
    		Session session; //会话
    		Destination dest; //消息的目的地
    		MessageConsumer consumer; //消费者
    		
    		try {
    			//初始化连接工厂
    			factory=new ActiveMQConnectionFactory(URL); 
    
    			//获取连接
    			conn=factory.createConnection(); 
    			
    			conn.setClientID("C1");
    			
    			//开始连接
    			conn.start(); 
    			
    			//创建Session，此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
    			session=conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			
    			//创建topic,producer和consumer将根据TOPIC_NAME来发送/接收对应的消息 
    			dest=session.createTopic(TOPIC_NAME); 
    			
    			//创建消费者
    			//consumer=session.createConsumer(dest);
    			
    			//创建消费者(持久化)
    			consumer = session.createDurableSubscriber((Topic) dest, TOPIC_NAME);
    			
    			//初始化MessageListener
    			MyConsumer me=new MyConsumer();
    			
    			//给消费者设定监听对象
    			consumer.setMessageListener(me);
    		} catch (Exception e) {
    			System.out.println("出错啦!!!");
    			e.printStackTrace();
    		}
    	}
    	
    	public void onMessage(Message message) {
    		TextMessage txtMessage = (TextMessage)message;
    		String text;
    		try {
    			text=txtMessage.getText();
    			//txtMessage.acknowledge(); //通知服务器
    			System.out.println("收到消息:"+text);
    		} catch (JMSException e) {
    			System.out.println("出错啦!!!");
    			e.printStackTrace();
    		}
    	}
    }

在这里我们可以编写两个消费者模拟群发的测试，须要对conn.setClient()设置不同的id。笔者使用了C1与C2，由于代码相同，我就不贴C2的代码了。

3.分析与测试

在使用jdbc的方式实现持久化的时候，activemq会在数据库创建三张表，ACTIVEMQ\_MSGS，ACTIVEMQ\_ACKS和ACTIVEMQ\_LOCK。第二张表外键关联到第一张表，共同存储消息，第三张表用于锁定保证只有一个broker实例可以访问数据库。

测试：

1.先启动MyConsumer类，让MyConsumer在自己的主题上进行订阅  
2.停止MyConsumer类，再启动MyProducer类  
3.待MyProducer类运行完成后，再启动MyConsumer类  
4.发现数据库中存在信息，相应主题的信息被订阅，证明持久化成功。