停止和重新启动 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设置为true，我不知道scala但这里是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;
}