在停止和重新启动时丢失消息Embedded ActiveMQ

Question

嗨我正在尝试编写一个测试用例来实现故障转移支持activeMQ。

下面是代码

val brokerA = createBroker("A") 
brokerA.start 
val failoverUrl = s"failover:(vm://BrokerA?create=false)" + 
s"?randomize=false&maxReconnectAttempts=-1&reconnectSupported=true" 


val cFactory = new ActiveMQConnectionFactory(failoverUrl) 
val qConnection = getQueueConnection 
val session = createQueueSession(qConnection) 

private def totalReadMessagesCount(queueReceiver: QueueReceiver) = { 
val messages = Iterator.continually(Option(queueReceiver.receive(2000))).takeWhile(_.isDefined).flatten.toSeq 
messages.size 
} 

private def getReceiver = { 
val queueConnection = getQueueConnection 
queueConnection.start() 
val queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE) 
val queueReceiver = createQueueReceiver(queueSession, brokerA.getBrokerName) 
queueReceiver 
} 

def getQueueConnection =cFactory.createQueueConnection("admin", "") 

def createBroker(name:String) = { 
val broker = new BrokerService() 
val adaptor = new KahaDBPersistenceAdapter() 
broker.setBrokerName("Broker" + name) 
broker.addConnector(getBrokerUrl) 
broker.setPersistent(true) 
broker.setUseJmx(false) 
broker.setUseShutdownHook(false) 
broker 
} 

def getBrokerUrl = "tcp://localhost:0" 


val queueReceiver: QueueReceiver = getReceiver 
val messageCount = 500 
(1 to messageCount) map {count => 
    //Calling method to send message to ActiveMQ 
    if(count == 200){ 
    brokerA.stop() 
    brokerA.waitUntilStopped() 
    brokerA.start(true) 
    } 
} 
val totalCount = totalReadMessagesCount(queueReceiver) 
println(s"Read ${totalCount} messages") 
assert(totalCount == messageCount) 

我能够重新启动后和ActiveMQ重新连接，但totalCount显示的是300而不是500，似乎之前的消息都将丢失。但是，当我在非嵌入模式下运行相同的场景。我能够获取所有消息。

请帮助我如何防止在重新启动嵌入式活动mq时丢失任何消息。

Answer 1

你必须persistent设置为真，我不知道斯卡拉，但这里是Java代码

public BrokerService broker() throws Exception { 
    final BrokerService broker = new BrokerService(); 
    //broker.addConnector("tcp://localhost:61616"); 
    broker.addConnector("stomp://localhost:61613"); 
    broker.addConnector("vm://localhost"); 
    PersistenceAdapter persistenceAdapter = new KahaDBPersistenceAdapter(); 
    File dir = new File(System.getProperty("user.home") + File.separator + "kaha"); 
    if (!dir.exists()) { 
     dir.mkdirs(); 
    } 
    persistenceAdapter.setDirectory(dir); 
    broker.setPersistenceAdapter(persistenceAdapter); 
    broker.setPersistent(true); 
    return broker; 
}