ActiveMQ Artemis事务失败后消息恢复最佳实践及缓冲区问题

提供的Kotlin代码 #

val connectionFactory = ActiveMQConnectionFactory(url)
val connection = connectionFactory.createConnection()
val session = connection.createSession(true, Session.SESSION_TRANSACTED)
val destination = session.createQueue("example.queue")

val producer = session.createProducer(destination)

// Maintain a list of messages sent within the transaction
val messagesInTransaction = mutableListOf<Message>()

// Send a batch of messages within a transaction
val numMessagesToSend = 10
for (i in 1..numMessagesToSend) {
    val message = session.createTextMessage("Hello, World! ($i)")
    producer.send(message)
    messagesInTransaction.add(message)
}

// Commit the transaction
try {
    session.commit()
    println("Transaction committed successfully")
} catch (e: Exception) {
    // Roll back the transaction and retry sending the messages
    session.rollback()
    println("Transaction rolled back: $e")
    
    for (message in messagesInTransaction) {
        println("Retrying sending message: $message")
        producer.send(message)
    }
}

// Close the JMS resources
producer.close()
session.close()
connection.close()

问题1：这种用List记录事务消息并重发的方式是事务失败恢复的最佳实践吗？ #

当然不是，这种方式存在不少明显缺陷，完全称不上最佳实践：

• 内存占用风险：如果批量发送的消息数量多、体积大，把所有消息存在内存List里会直接耗尽JVM内存，甚至触发OOM。
• 消息丢失隐患：如果进程在事务回滚后、重发前崩溃，内存里的消息会全部丢失，没有任何恢复手段。
• 重复消费问题：重发时没处理幂等性，如果Broker实际已经收到部分消息但返回提交失败，重发会导致重复消息，消费端若未做去重处理会引发业务异常。
• 对象状态异常：JMS的Message对象绑定到当前会话，跨会话复用可能出现内部状态异常（比如已被标记为已发送），导致重发失败。

更可靠的实践方向：

1. 利用客户端内置重试：ActiveMQ客户端支持配置事务级别的重试策略（比如通过ActiveMQConnectionFactory的setRetryInterval、setMaxReconnectAttempts等方法），提交失败时自动重试整个事务流程，无需手动维护消息列表。
2. 本地持久化暂存：发送前先把消息内容（文本、序列化字节等）持久化到本地数据库或文件，事务提交成功后再删除本地记录；提交失败时从本地存储读取消息重发，确保消息不会丢失。
3. 强制实现幂等性：给每个消息添加唯一业务标识（比如全局UUID），消费端根据标识去重，彻底避免重复处理的问题。

问题2：回滚时发送端消息缓冲区的实现原理，以及恢复示例？ #

实现原理

当你使用ActiveMQ的事务会话（SESSION_TRANSACTED）时，调用producer.send()后，消息并不会立刻发送到Broker，而是被暂存到当前会话专属的内存缓冲区（属于事务上下文的一部分）。只有当你调用session.commit()，客户端才会把缓冲区里的所有消息批量发送给Broker；如果调用session.rollback()，客户端会直接清空这个事务缓冲区，标记这些消息无需发送。

需要注意的是：这个缓冲区是临时的，回滚后会被自动清理，客户端不会主动保留这些消息——你没法直接从缓冲区里恢复回滚后的消息，这也是示例代码要手动维护List的原因。

可靠恢复的替代实现（模拟持久化缓冲区）

如果你想避免手动维护内存List，同时实现可靠的事务失败恢复，可以结合本地持久化+客户端重试的方式，示例如下：

val connectionFactory = ActiveMQConnectionFactory(url).apply {
    // 配置客户端自动重试策略
    setRetryInterval(1000)
    setMaxReconnectAttempts(3)
}

val connection = connectionFactory.createConnection()
connection.start()
val session = connection.createSession(true, Session.SESSION_TRANSACTED)
val destination = session.createQueue("example.queue")
val producer = session.createProducer(destination)

// 模拟本地持久化存储（实际项目建议用数据库或本地文件）
val persistentMessageStore = mutableListOf<String>()

val numMessagesToSend = 10
for (i in 1..numMessagesToSend) {
    val messageContent = "Hello, World! ($i)"
    // 先将消息内容持久化到本地
    persistentMessageStore.add(messageContent)
    val message = session.createTextMessage(messageContent)
    producer.send(message)
}

try {
    session.commit()
    println("Transaction committed successfully")
    // 提交成功，清空本地存储
    persistentMessageStore.clear()
} catch (e: Exception) {
    session.rollback()
    println("Transaction rolled back: $e")
    
    // 从本地存储恢复消息并重新发送
    println("Restoring messages from local store...")
    for (content in persistentMessageStore) {
        val message = session.createTextMessage(content)
        producer.send(message)
    }
    // 重新提交事务
    session.commit()
    println("Retried messages committed successfully")
    persistentMessageStore.clear()
} finally {
    // 关闭资源
    producer.close()
    session.close()
    connection.close()
}

内容的提问来源于stack exchange，提问作者Achraf Dridi