ActiveMQ 持久化（文件），查询队列剩余消息数、出队数的实现
本人博客开始迁移，博客整个架构本身搭建及编码 http://www.cookqq.com/listBlog.action java
《ActiveMQ发消息和收消息》详细介绍了ActiveMQ发消息和收消息，消息保存在消息队列（queue）中，消息队列数据保存在计算机内存中，假如ActiveMQ服务器因为某些缘由忽然中止，那消息队列中内容还在吗？用事实说话吧，把ActiveMQ服务器中止，而后再看看ActiveMQ页面上的队列信息queue，如图：数据库
activemq_queue队列中的信息所有丢失了。为了解决这个问题，能够把队列数据持久化，分为文件持久化和数据库持久化。服务器
如今详细分析文件持久化。直接来代码吧：session
消息发送者：架构
package com.activemq.queue;
import javax.jms.Connection;
public class Sender {
private static final int SEND_NUMBER = 1;
public static void main(String[] args) {
// ConnectionFactory 接口（链接工厂） 用户用来建立到JMS提供者的链接的被管对象。
// JMS客户经过可移植的接口访问链接，这样当下层的实现改变时，代码不须要进行修改。
// 管理员在JNDI名字空间中配置链接工厂，这样，JMS客户才可以查找到它们。
// 根据消息类型的不一样，用户将使用队列链接工厂，或者主题链接工厂。
ConnectionFactory connectionFactory;
// Connection ：JMS 客户端到JMS Provider 的链接
// Connection 接口（链接） 链接表明了应用程序和消息服务器之间的通讯链路。
// 在得到了链接工厂后，就能够建立一个与JMS提供者的链接。根据不一样的链接类型，
// 链接容许用户建立会话，以发送和接收队列和主题到目标。
Connection connection = null;
// Session： 一个发送或接收消息的线程
// Session 接口（会话） 表示一个单线程的上下文，用于发送和接收消息。
// 因为会话是单线程的，因此消息是连续的，就是说消息是按照发送的顺序一个一个接收的。
// 会话的好处是它支持事务。若是用户选择了事务支持，会话上下文将保存一组消息，直到事务被提交才发送这些消息。
// 在提交事务以前，用户能够使用回滚操做取消这些消息。一个会话容许用户建立消息生产者来发送消息，建立消息消费者来接收消息。
Session session;
// Destination ：消息的目的地;消息发送给谁.
// Destination 接口（目标） 目标是一个包装了消息目标标识符的被管对象，
// 消息目标是指消息发布和接收的地点，或者是队列，或者是主题。
// JMS管理员建立这些对象，而后用户经过JNDI发现它们。
// 和链接工厂同样，管理员能够建立两种类型的目标，点对点模型的队列，以及发布者/订阅者模型的主题。
Destination destination;
// MessageProducer：消息发送者
// MessageProducer 接口（消息生产者） 由会话建立的对象，用于发送消息到目标。
// 用户能够建立某个目标的发送者，也能够建立一个通用的发送者，在发送消息时指定目标。
MessageProducer producer;
// TextMessage message;
// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER, //null
ActiveMQConnection.DEFAULT_PASSWORD, //null
"tcp://localhost:61616");
try {
// 构造从工厂获得链接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操做链接
session = connection.createSession(Boolean.TRUE,
Session.AUTO_ACKNOWLEDGE); //1
destination = session.createQueue(RunServer.queueName);
// 获得消息生成者【发送者】
producer = session.createProducer(destination);
// 设置不持久化，能够更改
producer.setDeliveryMode(DeliveryMode.PERSISTENT); //2
// 构造消息
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
public static void sendMessage(Session session, MessageProducer producer)
throws Exception {
for (int i = 1; i <= SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq 发送的消息" + i);
// 发送消息到目的地方
System.out.println("发送消息:" + i);
producer.send(message);
}
}
}
消息接受者：异步
package com.activemq.queue;
import javax.jms.Connection;
public class Receiver {
public static void main(String[] args) {
// ConnectionFactory ：链接工厂，JMS 用它建立链接
ConnectionFactory connectionFactory;
// Connection ：JMS 客户端到JMS Provider 的链接
Connection connection = null;
// Session： 一个发送或接收消息的线程
final Session session;
// Destination ：消息的目的地;消息发送给谁.
Destination destination;
// 消费者，消息接收者
// MessageConsumer 接口（消息消费者） 由会话建立的对象，用于接收发送到目标的消息。
// 消费者能够同步地（阻塞模式），或异步（非阻塞）接收队列和主题类型的消息。
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://localhost:61616");
try {
// 构造从工厂获得链接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操做链接
session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
//queue_name跟sender的保持一致，一个建立一个来接收
destination = session.createQueue(RunServer.queueName);
consumer = session.createConsumer(destination);
// //第一种状况
// int i = 0;
// while (i < 3) {
// i++;
// TextMessage message = (TextMessage) consumer.receive();
// session.commit();
// // TODO something....
// System.out
// .println("收到消息：" +message.getText());
// }
// session.close();
// connection.close();
//----------------第一种状况结束----------------------
// 第二种方式
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message arg0) {
if(arg0 instanceof TextMessage){
try {
System.out.println("arg0="+((TextMessage)arg0).getText());
session.commit();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
//第三种状况
// while (true) {
// Message msg = consumer.receive(1000);
// TextMessage message = (TextMessage) msg;
// if (null != message) {
// System.out.println("收到消息:" + message.getText());
// }
// }
} catch (Exception e) {
e.printStackTrace();
}
}
}
程序中启动activemq服务：tcp
package com.activemq.queue;
import java.io.File;
public class RunServer {
public static String jmxDomain = "jms-broker";
public static int connectorPort = 2011;
public static String connectorPath = "/jmxrmi";
public static String queueName = "activemq_queue";
/** 启动activeMQ服务 */
public static void main(String[] args) throws Exception {
// java代码调用activemq相关的类来构造并启动brokerService
BrokerService broker = new BrokerService();
// 如下是持久化的配置
// 持久化文件存储位置
File dataFilterDir = new File("activemq/amq-in-action/kahadb");
KahaDBStore kaha = new KahaDBStore();
kaha.setDirectory(dataFilterDir);
// use a bigger journal file
kaha.setJournalMaxFileLength(1024*100);
// small batch means more frequent and smaller writes
kaha.setIndexWriteBatchSize(100);
// do the index write in a separate thread
kaha.setEnableIndexWriteAsync(true);
broker.setPersistenceAdapter(kaha);
// create a transport connector
broker.addConnector("tcp://localhost:61616");
broker.setUseJmx(true);
// 如下是ManagementContext的配置，从这个容器中能够取得消息队列中未执行的消息数、消费者数、出队数等等
// 设置ManagementContext
ManagementContext context = broker.getManagementContext();
context.setConnectorPort(connectorPort);
context.setJmxDomainName(jmxDomain);
context.setConnectorPath(connectorPath);
broker.start();
}
}
测试队列queue中各个状态数：ide
package com.activemq.queue;
import javax.management.MBeanServerConnection;
public class StateTest {
/**
* 获取状态
* @throws Exception
*/
public static void main(String[] args) throws Exception {
JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:"+
RunServer.connectorPort+RunServer.connectorPath);
JMXConnector connector = JMXConnectorFactory.connect(url, null);
connector.connect();
MBeanServerConnection connection = connector.getMBeanServerConnection();
// 须要注意的是，这里的jms-broker必须和上面配置的名称相同
ObjectName name = new ObjectName(RunServer.jmxDomain+":BrokerName=localhost,Type=Broker");
BrokerViewMBean mBean = (BrokerViewMBean)MBeanServerInvocationHandler.newProxyInstance(connection,
name, BrokerViewMBean.class, true);
// System.out.println(mBean.getBrokerName());
for(ObjectName queueName : mBean.getQueues()) {
QueueViewMBean queueMBean = (QueueViewMBean)MBeanServerInvocationHandler
.newProxyInstance(connection, queueName, QueueViewMBean.class, true);
System.out.println("\n------------------------------\n");
// 消息队列名称
System.out.println("States for queue --- " + queueMBean.getName());
// 队列中剩余的消息数
System.out.println("Size --- " + queueMBean.getQueueSize());
// 消费者数
System.out.println("Number of consumers --- " + queueMBean.getConsumerCount());
// 出队数
System.out.println("Number of dequeue ---" + queueMBean.getDequeueCount() );
}
}
}