测试方式:
使用我们之前使用的脚本,在hornetq做failover的环境下,施加很大的压力(50个线程),看failover能否成功(看有没有丢数据,主-副机能不能正常的切换过来)
具体的操作方式是:
Hornetq自带的example有HA这块的测试脚本
/hornetq-2.1.2.Final/examples/jms/non-transaction-failover
/hornetq-2.1.2.Final/examples/jms/transaction-failover
执行[bes@test157 transaction-failover]$./build.sh
带broker起来后启动压力测试脚本
需要修改的配置
hornetq-jms.xml
增加对应的destination(为我们压力脚本设置的目的地)
<topic name="topic1">
<entry name="/my/Topic1"/>
</topic>
hornetq-configuration.xml
在<configuration>节点下增加
<security-enabled>false</security-enabled>
hornetq-beans.xml和client-jndi.properties中对应的localhost替换为本机的ip
增加两个对应的topic的测试类,在原有的用例中没有topic的测试用例
日志数据的比较:
Hornetq日志数据的比较调用堆栈:
ReplicationCompareDataMessage
-decode:PacketDecoder
--bufferReceived:RemotingConnectionImpl
----bufferReceived:DelegatingBufferHandler
--------bufferReceived:DelegatingBufferHandler
日志比较的关键代码org.hornetq.core.replication.impl.ReplicationManagerImpl
public void compareJournalInformation(
final JournalLoadInformation[] journalInformation) throws
HornetQException
{
if (journalLoadInformation == null ||
journalLoadInformation.length != journalInformation.length)
{
throw new HornetQException(
HornetQException.INTERNAL_ERROR,
"Live Node contains more journals than the backup node. Probably a version match error");
}
for (int i = 0; i < journalInformation.length; i++)
{
if (!journalInformation[i].equals
(journalLoadInformation[i])) {
ReplicationEndpointImpl.log
.warn("Journal comparission mismatch:\n" +
journalParametersToString(journalInformation));
}
}
}
这里面重写了org.hornetq.core.journal.JournalLoadInformation的equals方法,在看看它的equals方法
JournalLoadInformation other = (JournalLoadInformation) obj;
if(maxID != other.maxID){
return false;
}
if(numberOfRecords != other.numberOfRecords){
return false;
}
return true;
从上面我们可以看出,它比较的是maxID以及numberOfRecords这两个值。我们在看看其中的一个赋值的地方:
public void decodeRest(final HornetQBuffer buffer) {
int numberOfJournals = buffer.readInt();
journalInformation = new JournalLoadInformation[numberOfJournals];
for (int i = 0; i < numberOfJournals; i++) {
journalInformation[i] = new JournalLoadInformation();
journalInformation[i].setNumberOfRecords(buffer.readInt());
journalInformation[i].setMaxID(buffer.readLong());
}
}
就目前的调查来看,在启动HornetQ和创建session会话的时候会调用到日志比较。下图是比较日志调用的路线
compareJournals(ReplicationManagerImpl)
compareJournals(HornetQServerImpl)
initialisePart2(HornetQServerImpl)
checkActivate(HornetQServerImpl)
handleCreateSession(HornetQPacketHandler)
handleReattachSession(HornetQPacketHandler)
start(HornetQServerImpl)
日志复制通道
RepliactionManagerImpl#start方法用于获取一个与备原机器的连接,创建用于日志复制的会话
:
start(ReplicationManagerImpl)
activated(JMSServerMangerImpl)
createJournal(JMSServerMangerImpl)
initJournal(JMSServerMangerImpl)
activated(JMSServerMangerImpl)
callActivatedCallbacks(HornetQServerImpl)
initialisePart2(HornetQServerImpl)
checkActivate(HornetQServerImpl)
start(HornetQServerImpl)
关键代码org.hornetq.core.replication.impl.ReplicationManagerImpl#start
public synchronized void start() throws Exception {
// 获取和备原机器的连接
replicatingConnection = failoverManager.getConnection();
long channelID = replicatingConnection.generateChannelID();
Channel mainChannel = replicatingConnection.getChannel(1, -1);
replicatingChannel = replicatingConnection.getChannel(channelID, -1);
replicatingChannel.setHandler(responseHandler);
CreateReplicationSessionMessage replicationStartPackage = new CreateReplicationSessionMessage(
channelID);
// 发送一个创建拷贝会话的命令(PacketImpl.CREATE_REPLICATION)
mainChannel.sendBlocking(replicationStartPackage);
}
上面发出的消息HorentQPacketHanler#handlePacket(final Packet packet)会处理
case CREATE_REPLICATION: {
//Create queue can also be fielded here in the case of a replicated store and forward queue creation
CreateRelicationSessionMessage request = (CreateRelicationSessionMessage)packet;
handleCreateReplication(request);
break;
}
日志的同步
调用ReplicationManagerImpl#sendReplicatePacket来复制日志;这个方法的调用者很多,如消息的发送,结束发送,创建连接工厂,物理目的地等操作
sendReplicatePacket(ReplicationManagerImpl)
appendAddRecord(ReplicatedJournal)
appendAddRecord(ReplicatedJournal)
storeMessage(JournalStroageManager)
processRoute(PostOfficeImpl)
redistribute(PostOfficeImpl)
route(PostOfficeImpl)
routeQueueInfo(PostOfficeImpl)
关键代码org.hornetq.core.replication.impl.ReplicationManagerImpl#sendReplicationPacket
private void sendReplicatePacket(final Pakcet packet)
{
boolean runItNow = false;
OperationContext repliToken = OperationContextImpl
.getContext(executorFactory);
repliToken.replicationLineUp();
synchronized (replicationLock)
{
if (!enabled)
{
runItNow = true;
}
else
{
pendingTokens.add(repliToken);
replicatingChannel.send(packet);
}
}
if (runItNow)
{
repliToken.relicationDone();
}
}
收到一条消息就发到replicatingChannel,做到了日志同步。
replicatingChannel的建立
org.hornetq.core.replication.impl.ReplicationManagerImpl#start
public synchronized void start() throws Exception {
replicatingConnection = failoverManager.getConnection();
long channelID = replicatingConnection.generateChannelID();
//在这里为replicatingChannel赋值
replicatingChannel = replicatingConnection.getChannel(channelID, -1);
replicatingChanne.setHandler(responseHandler);
}
主备机器切换
HornetQ的失效备原是在客户端层面来做的
,之前会注册一个用于失效备原的监听器,当监听到异常时,就会尝试进行失效备原;下面是在发生失效备原时的调用堆栈:
FailoverManagerImpl#failoverOrReconnect
FailoverManagerImpl#handleConnectionFailure
FailoverManagerImpl$DelegatingFailureListener#connectionFailed
RemotingConnectionImpl#callFailureListeners
RemotingConnectionImpl#fail
FailoverManagerImpl$ChannelOHandler
Failover关键代码org.hornetq.core.client.impl.FailoverManagerImpl#failoverOrReconnect
分享到:
相关推荐
教你一步一步掌握HornetQ,HornetQ是一个支持集群和多种协议,可嵌入、高性能的异步消息系统。HornetQ完全支持JMS,HornetQ不但支持JMS1.1 API同时也定义属于自己的消息API,这可以最大限度的提升HornetQ的性能和...
HornetQ 2.3.0 Alpha 发布,这不是一个简单的 Alpha 版本,同时也是一个大的发行版本。该版本对 2.2.0 进行了重构,引入一些原子故障迁移特性和大量企业特性改进。详细的新特性介绍请看发行说明。 HornetQ是一个...
HornetQ是JBoss社区所研发的开放源代码消息中间件;HornetQ是以Java 5 编写,因此只需要操作系统支持Java虚拟机,HornetQ便可运行。 支持Java消息服务 (JMS) 1.1 版本 集群 (Clustering) 支持庞大的消息(Message)...
Hornetq 实例
HornetQ2.1中文手册 HornetQ2.1中文手册
hermes 监听hornetq JMS配置
该文档是对Hornetq英文文档的翻译,对于英文不好的同学,有很大的帮助。
hornetq 2.4.0 一个轻量但高效的MQ
ActiveMQ和HornetQ性能对比
hornetq-2.2.5.Final.ziphornetq-2.2.5.Final.ziphornetq-2.2.5.Final.ziphornetq-2.2.5.Final.ziphornetq-2.2.5.Final.ziphornetq-2.2.5.Final.ziphornetq-2.2.5.Final.zip
能用该dll连HornetQ并上传或获取队列数据,引用dll,创建Factory ,创建连接,创建消费者,创建发布者,等等
hornetq安装包, hornetq-2.4.0.Final-bin.tar 消息中间件 供项目中数据交互使用
hornetq-2.3.0.Final-bin.zip
HornetQ2.1中文手册.7z
HornetQ提供自动客户端失效备援(automatic client failover)功能,能保证在服务器故障时没有消息丢失或消息重复。 * 超级灵活的集群方案。可以控制集群进行消息负载均衡的方式。分布在不同地理位置的各个集群间...
hornetq-transports-2.0.0.GA.jar
HornetQ集群配置,主主集群,通过udp广播形式通信。。
HornetQ是java开源实现的消息系统框架,性能上比ActiveQ要好一些,被集成到JBoss的消息服务中。 Table of Contents Preface 1 Chapter 1: Getting Started with HornetQ 9 Chapter 2: Setting Up HornetQ 31 ...
hornetQ 2.4.0
Sumac.zip,scala中的sumac ext zkargument解析