AMQ消息持久 化
(2)数据库表信息
activemq_msgs用于存储消息,Queue和Topic都存储在这个表中:
ID:自增的数据库主键
CONTAINER:消息的Destination
MSGID_PROD:消息发送者客户端的主键
MSG_SEQ:是发送消息的顺序,MSGID_PROD+MSG_SEQ可以组成JMS的MessageID
EXPIRATION:消息的过期时间,存储的是从1970-01-01到现在的毫秒数
MSG:消息本体的Java序列化对象的二进制数据
PRIORITY:优先级,从0-9,数值越大优先级越高
activemq_acks用于存储订阅关系。如果是持久化Topic,订阅者和服务器的订阅关系在这个表保存:
主要的数据库字段如下:
CONTAINER:消息的Destination
SUB_DEST:如果是使用Static集群,这个字段会有集群其他系统的信息
CLIENT_ID:每个订阅者都必须有一个唯一的客户端ID用以区分
SUB_NAME:订阅者名称
SELECTOR:选择器,可以选择只消费满足条件的消息。条件可以用自定义属性实现,可支持多属性AND和OR操作
LAST_ACKED_ID:记录消费过的消息的ID。
表activemq_lock在集群环境中才有用,只有一个Broker可以获得消息,称为Master Broker,
其他的只能作为备份等待Master Broker不可用,才可能成为下一个Master Broker。
这个表用于记录哪个Broker是当前的Master Broker。
2. AMQ方式
性能高于JDBC,写入消息时,会将消息写入日志文件,由于是顺序追加写,性能很高。为了提升性能,创建消息主键索引,并且提供缓存机制,进一步提升性能。每个日志文件的大小都是有限制的(默认32m,可自行配置)。
当超过这个大小,系统会重新建立一个文件。当所有的消息都消费完成,系统会删除这个文件或者归档(取决于配置)。
主要的缺点是AMQ Message会为每一个Destination创建一个索引,如果使用了大量的Queue,索引文件的大小会占用很多磁盘空间。
而且由于索引巨大,一旦Broker崩溃,重建索引的速度会非常慢。
配置片段如下:
1 2 3 |
|
虽然AMQ性能略高于下面的Kaha DB方式,但是由于其重建索引时间过长,而且索引文件占用磁盘空间过大,所以已经不推荐使用。
3. KahaDB方式
KahaDB是从ActiveMQ 5.4开始默认的持久化插件,也是我们项目现在使用的持久化方式。
KahaDb恢复时间远远小于其前身AMQ并且使用更少的数据文件,所以可以完全代替AMQ。
kahaDB的持久化机制同样是基于日志文件,索引和缓存。
配置方式:
1 2 3 4 5 |
|
(1)KahaDB主要特性
1、日志形式存储消息;
2、消息索引以B-Tree结构存储,可以快速更新;
3、完全支持JMS事务;
4、支持多种恢复机制;
(2)KahaDB的结构
消息存储在基于文件的数据日志中。如果消息发送成功,变标记为可删除的。系统会周期性的清除或者归档日志文件。
消息文件的位置索引存储在内存中,这样能快速定位到。定期将内存中的消息索引保存到metadata store中,避免大量消息未发送时,消息索引占用过多内存空间。
Data logs:
Data logs用于存储消息日志,消息的全部内容都在Data logs中。
同AMQ一样,一个Data logs文件大小超过规定的最大值,会新建一个文件。同样是文件尾部追加,写入性能很快。
每个消息在Data logs中有计数引用,所以当一个文件里所有的消息都不需要了,系统会自动删除文件或放入归档文件夹。
Metadata cache :
缓存用于存放在线消费者的消息。如果消费者已经快速的消费完成,那么这些消息就不需要再写入磁盘了。
Btree索引会根据MessageID创建索引,用于快速的查找消息。这个索引同样维护持久化订阅者与Destination的关系,以及每个消费者消费消息的指针。
Metadata store
在db.data文件中保存消息日志中消息的元数据,也是以B-Tree结构存储的,定时从Metadata cache更新数据。Metadata store中也会备份一些在消息日志中存在的信息,这样可以让Broker实例快速启动。
即便metadata store文件被破坏或者误删除了。broker可以读取Data logs恢复过来,只是速度会相对较慢些。
4.LevelDB方式
从ActiveMQ 5.6版本之后,又推出了LevelDB的持久化引擎。
目前默认的持久化方式仍然是KahaDB,不过LevelDB持久化性能高于KahaDB,可能是以后的趋势。
在ActiveMQ 5.9版本提供了基于LevelDB和Zookeeper的数据复制方式,用于Master-slave方式的首选数据复制方案。
如在实例在mysql序列化时,
public class Producer {
public static void main(String[] args) {
// String user = ActiveMQConnection.DEFAULT_USER;
// String password = ActiveMQConnection.DEFAULT_PASSWORD;
// String url = ActiveMQConnection.DEFAULT_BROKER_URL;
String subject = "TestQueue";
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("wjq008wang", "123456", "failover:(tcp://localhost:61616)?Randomize=false");
// ConnectionFactory contectionFactory = new ActiveMQConnectionFactory("tcp://192.168.0.207:61616");
// ConnectionFactory contectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
try{
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(subject);
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);//设置为持久化
for(int i = 0; i < 20;) {
TextMessage createTextMessage = session.createTextMessage("这是要发送的第"+ ++i +"条消息消息");
producer.send(createTextMessage);
System.out.println("第"+ i +"条消息已发送");
}
Thread.sleep(2000);
session.commit();
session.close();
connection.close();
}catch (JMSException e) {
// e.printStackTrace();
}catch (InterruptedException e) {
// e.printStackTrace();
}
}
}
消费者
package example;
/**
* @Header: Receiver.java
* 类描述:
* @author: lm
* @date 2013-7-17 上午10:52:58
* @Email
* @company 欢
* @addr 北京市朝阳区劲松
*/
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Receiver {
public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话
Destination destination; // 消息地址
MessageConsumer messageConsumer; // 消息消费者
// 实例化连接工厂
connectionFactory = new ActiveMQConnectionFactory("wjq008wang", "123456", "failover:(tcp://localhost:61616)?Randomize=false");
try {
// 创建连接
connection = connectionFactory.createConnection();
// 连接开启
connection.start();
// 创建Session会话
// 参数1:是否开启事务
// 参数2:会话Session
// Session.AUTO_ACKNOWLEDGE - 自动确认 消费者从receive或监听成功返回时,自动确认客户端收到消息
// Session.CLIENT_ACKNOWLEDGE - 客户通过acknowledge方法确认消息(会话层确认),确认一个即确认所有
// Session.DUPS_OK_ACKNOWLEDGE - 重复确认
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 创建连接消息队列
destination = session.createQueue("TestQueue");
// 创建消息消费者
messageConsumer = session.createConsumer(destination);
// 接收数据-Receive方式-由于一直要轮训接收,不推荐
while(true){
TextMessage textMessage= (TextMessage)messageConsumer.receive(10000);//每10000毫秒接收一次消息
if(textMessage != null){
System.out.println("消费者-Receive方式-收到消息: " + textMessage.getText());
}
}
} catch (Exception e) {
e.printStackTrace();
System.out.println("错误类名称 = " + e.getClass().getName());
System.out.println("错误原因 = " + e.getMessage());
}finally { //关闭释放
if(connection != null){
try {
connection.close();
connection = null;
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
}