文档章节

ActiveMQ 持久化(数据库),查询队列剩余消息数、出队数的实现

cookqq
 cookqq
发布于 2013/03/04 09:57
字数 754
阅读 4.6K
收藏 24

#程序员薪资揭榜#你做程序员几年了?月薪多少?发量还在么?>>>

《ActiveMQ 持久化(文件),查询队列剩余消息数、出队数的实现》分析了消息队列持久化保存,假如activemq服务器突然停止,服务器启动后,还可以继续查找队列中的消息。现在分析队列中的消息使用数据库持久化。

本人博客开始迁移,博客整个架构自己搭建及编码http://www.cookqq.com/

消息生产者:

package com.activemq.mysql;

import java.io.File;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.sql.DataSource;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
import org.apache.activemq.store.jdbc.adapter.MySqlJDBCAdapter;
import org.apache.commons.dbcp.BasicDataSourceFactory;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

/**
 * 消息持久化到数据库
 *
 */
public class MessageProductor {
	  private static Logger logger=LogManager.getLogger(MessageProductor.class);
	  private String username=ActiveMQConnectionFactory.DEFAULT_USER;
	  private String password=ActiveMQConnectionFactory.DEFAULT_PASSWORD;
	  private  String url=ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL;
	  
	  public static String queueName="acticemq_queue";
	  private BrokerService brokerService;
	  protected static final int messagesExpected = 3;
	  
	  protected ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
			    username,password,
	            "tcp://localhost:61616?jms.prefetchPolicy.all=0&jms.redeliveryPolicy.maximumRedeliveries="+messagesExpected);
	/***
	 * 创建Broker服务对象
	 * @return
	 * @throws Exception
	 */
	public BrokerService createBroker()throws Exception{
			BrokerService  broker=new BrokerService();
			JDBCPersistenceAdapter jdbc=createJDBCPersistenceAdapter();
			broker.setPersistenceAdapter(jdbc);
			jdbc.setDataDirectory(System.getProperty("user")+
					File.separator+"data"+File.separator);
			jdbc.setAdapter(new MySqlJDBCAdapter());
			broker.setPersistent(true);
			broker.addConnector("tcp://localhost:61616");
			//broker.addConnector(ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL);
		return broker;
	}
	/**
	 * 创建Broken的持久化适配器
	 * @return
	 * @throws Exception
	 */
	public JDBCPersistenceAdapter createJDBCPersistenceAdapter() throws Exception{
		JDBCPersistenceAdapter jdbc=new JDBCPersistenceAdapter();
		DataSource datasource=createDataSource();
		jdbc.setDataSource(datasource);
		jdbc.setUseDatabaseLock(false);
		//jdbc.deleteAllMessages();
		return jdbc;
	}
	/**
	 * 创建数据源
	 * @return
	 * @throws Exception
	 */
	public DataSource createDataSource() throws Exception{
		Properties props=new Properties();
		props.put("driverClassName", "com.mysql.jdbc.Driver");
		props.put("url", "jdbc:mysql://localhost:3306/activemq");
		props.put("username", "root");
		props.put("password", "16ds");
		DataSource datasource=BasicDataSourceFactory.createDataSource(props);
		return datasource;
	}
	/**
	 * 启动BrokerService进程
	 * @throws Exception
	 */
	public void init() throws Exception{
		createBrokerService();
		start();
	}
	
	public void start() throws Exception{
		if(brokerService!=null){
			brokerService.start();
		}
	}
	public BrokerService createBrokerService() throws Exception{
		if(brokerService==null){
			brokerService=createBroker();
		}
		return brokerService;
	}
	
	public void sendMessage() throws JMSException{
		Connection connection=connectionFactory.createConnection();
		connection.start();
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	    Destination destination = session.createQueue(queueName);        
	    MessageProducer producer = session.createProducer(destination);
	    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
		for(int i=0;i<messagesExpected;i++){
			 logger.debug("Sending message " + (i+1) + " of " + messagesExpected);
	         producer.send(session.createTextMessage("test message " + (i+1)));
		}
		connection.close();
	}
	public String getUrl() {
		return url;
	}
	public void setUrl(String url) {
		this.url = url;
	}
	public String getUsername() {
		return username;
	}
	public void setUsername(String username) {
		this.username = username;
	}
}

消息消费者:

package com.activemq.mysql;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
/***
 * 消息持久化到数据库
 */
public class MessageCustomer {
	private static Logger logger=LogManager.getLogger(MessageProductor.class);
	  protected static final int messagesExpected = 5;
	  
	/***
	 * 创建Broker服务对象
	 * @return
	 * @throws Exception
	 */
	public BrokerService createBroker()throws Exception{
		BrokerService  broker=new BrokerService();
	    broker.addConnector(ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL);
		return broker;
	}

	/**
	 * 启动BrokerService进程
	 * @throws Exception
	 */
	public void init() throws Exception{
		BrokerService brokerService=createBroker();
		brokerService.start();
	}
	/**
	 * 接收的信息
	 * @return
	 * @throws Exception
	 */
	public int receiveMessage() throws Exception{
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
	            "tcp://localhost:61616?jms.prefetchPolicy.all=0&jms.redeliveryPolicy.maximumRedeliveries="+messagesExpected);
		Connection connection=connectionFactory.createConnection();
		connection.start();
		Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
		return receiveMessages(messagesExpected,session);
	}
	

	/**
	 * 接受信息的方法
	 * @param messagesExpected
	 * @param session
	 * @return
	 * @throws Exception
	 */
	protected int receiveMessages(int messagesExpected, Session session) throws Exception {
        int messagesReceived = 0;
        for (int i=0; i<messagesExpected; i++) {
            Destination destination = session.createQueue(MessageProductor.queueName);
            MessageConsumer consumer = session.createConsumer(destination);
            Message message = null;
            try {
            	logger.debug("Receiving message " + (messagesReceived+1) + " of " + messagesExpected);
                message = consumer.receive(2000);
                logger.info("Received : " + message);
                System.out.println("Received : " + message);
                if (message != null) {
                    session.commit();
                    messagesReceived++;
                }
            } catch (Exception e) {
            	logger.debug("Caught exception " + e);
                session.rollback();
            } finally {
                if (consumer != null) {
                    consumer.close();
                }
            }
        }
        return messagesReceived;
    }

}

生产者测试类:

package com.activemq.mysql;

public class MessageProductorTest {
	
	public static void main(String[] args) throws Exception {
		MessageProductor  productor =new MessageProductor();
		productor.init();
		productor.sendMessage();
		//productor.createBrokerService().stop();
	}

}

消费者测试类:

package com.activemq.mysql;


public class MessageCustomerTest {
  public static void main(String[] args) throws Exception {
	  MessageCustomer  customer=new MessageCustomer();
	  //customer.init();  //当两台机器在不同的服务器上启动客户端的broker进程
	  customer.receiveMessage();
	  
}
}

数据库形式:

activemq_acks:ActiveMQ的签收信息。

activemq_lock:ActiveMQ的锁信息。

activemq_msgs:ActiveMQ的消息的信息



参照博客: http://topmanopensource.iteye.com/blog/1066383

© 著作权归作者所有

cookqq

cookqq

粉丝 119
博文 268
码字总数 156096
作品 0
海淀
技术主管
私信 提问
加载中

评论(0)

ActiveMQ发消息、收消息、持久化,查询队列剩余消息数、出队数的实现

初次发博文,勿喷~~ 最近老大让我使用ActiveMQ实现这么个东东:1.查询消息队列中还有多少任务没有执行;2.消息队列的持久化; 真是愁杀我也,以前没见过啊,于是又看文档,又百度又google的,...

JingHaiChao
2012/05/14
1W
7
ActiveMQ 持久化(文件),查询队列剩余消息数、出队数的实现

本人博客开始迁移,博客整个架构自己搭建及编码 http://www.cookqq.com/listBlog.action 《ActiveMQ发消息和收消息》详细介绍了ActiveMQ发消息和收消息,消息保存在消息队列(queue)中,消息...

cookqq
2013/03/03
1.3W
1
ActiveMq笔记2-消息持久化

为了避免意外宕机以后丢失信息,需要做到重启后可以恢复消息队列,消息系统一般都会采用持久化机制。 ActiveMQ的消息持久化机制有JDBC,AMQ,KahaDB和LevelDB, 无论使用哪种持久化方式,消息...

狂小白
2018/02/25
0
0
MQ选型对比RabbitMQ RocketMQ ActiveMQ

原文:MQ选型对比RabbitMQ RocketMQ ActiveMQ 几种MQ产品说明: ZeroMQ : 扩展性好,开发比较灵活,采用C语言实现,实际上他只是一个socket库的重新封装,如果我们做为消息队列使用,需要开发...

osc_51w4wn2j
2019/01/29
15
0
消息中间件系列第2讲:如何进行消息队列选型?

要做技术选型,那么必须对现今的各个消息中间件有个深入的理解才能做技术选型。否则别人问你,你为什么要用这个消息中间件,你说不出个所以然来,怎么做架构师呢? 截止到目前为止,现在业界...

陈树义
2019/01/07
0
0

没有更多内容

加载失败,请刷新页面

加载更多

centos7安装squid代理

局域网只有一台服务器可以上互联网,其他机器需要使用代理上网,windows下可以用ccproxy,linux建议使用squid(dns解析需要配合iptables) 1、安装squid yum install squid.x86_64 2、配置squ...

osc_rn23gf4h
8分钟前
8
0
基于表单的网站身份验证的权威指南[关闭] - The definitive guide to form-based website authentication [closed]

问题: Form-based authentication for websites 基于表单的网站身份验证 We believe that Stack Overflow should not just be a resource for very specific technical questions, but also......

fyin1314
9分钟前
7
0
我的第一个Flask项目

项目背景提要 最近公司经常有测试,产品,开发人员需要我帮忙查看服务器上面发送的短信验证码来完成工作上的一些需求。我们的短信验证码由我们后台程序发出,调用第三方短信平台发送,这中间...

osc_6kvl6c8h
10分钟前
10
0
云原生下的开发测试之困与阿里的解决之道

【以下为分享实录,有删节】 测试环境管理之困与阿里巴巴的解决之道 在云原生时代下,软件的迭代速度越来越快,对测试的要求也越来越高,很多开发者开始使用Kubernetes来管理测试环境。在这个...

阿里云技术博客
10分钟前
5
0
Active Directory颗粒化密码策略配置

1 多元(颗粒化)密码策略介绍 在windows server 2000/2003中,我们无法针对域用户不同而设置不同密码策略, 域用户密码策略和账户设置都 由默认域策略控制,如果要重新建立策略我们必须创建...

osc_61i1fz2h
11分钟前
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部