使用akka.persistence.postgresql遇too many clients异常的最优解决方案咨询

1. 配置Akka Persistence PostgreSQL的连接池 #

Akka Persistence PostgreSQL默认的连接管理可能未充分利用连接池，导致每个Actor尝试创建独立连接。通过配置文件强制使用连接池并复用连接，可从根源减少连接占用：

在application.conf中添加如下配置：

akka.persistence.journal.postgresql {
  connection-pool = "HikariCP"
  hikaricp {
    maximum-pool-size = 20  # 根据服务器承载能力调整，远小于PostgreSQL的max_connections
    minimum-idle = 5
    idle-timeout = 30000
    connection-timeout = 20000
  }
}

akka.persistence.snapshot-store.postgresql {
  connection-pool = "HikariCP"
  hikaricp {
    # 可与journal共享连接池，或单独配置，建议共享以控制总连接数
    maximum-pool-size = 10
  }
}

所有Actor的持久化操作都会复用池中的连接，避免每个Actor占用独立连接。

2. 共享持久化上下文 #

通过Akka Extension机制，让多个Actor共享同一个持久化上下文实例，避免重复初始化连接资源：

import akka.actor.ExtendedActorSystem
import akka.actor.Extension
import akka.actor.ExtensionId
import akka.actor.ExtensionIdProvider
import akka.persistence.journal.JournalPersistenceApi
import akka.persistence.postgresql.journal.PostgreSQLJournal

class SharedPersistenceContext(system: ExtendedActorSystem) extends Extension {
  val journalApi: JournalPersistenceApi = PostgreSQLJournal(system)
}

object SharedPersistenceContext extends ExtensionId[SharedPersistenceContext] with ExtensionIdProvider {
  override def createExtension(system: ExtendedActorSystem): SharedPersistenceContext =
    new SharedPersistenceContext(system)

  override def lookup(): ExtensionId[_ <: Extension] = this
}

在PersistentActor中通过Extension获取共享的journal API：

class MyPersistentActor extends PersistentActor {
  private val sharedJournal = SharedPersistenceContext(context.system).journalApi

  override def persistenceId: String = "my-actor-id"

  override def receiveCommand: Receive = {
    case evt: Event =>
      sharedJournal.persist(evt) { _ =>
        // 业务处理逻辑
      }
  }
}

3. 批量持久化事件 #

缓存多个事件后批量提交，减少单条事件的持久化请求，降低连接使用频率：

class BatchPersistentActor extends PersistentActor {
  private val batchSize = 50
  private var eventBuffer = List.empty[Event]

  override def persistenceId: String = "batch-actor-id"

  override def receiveCommand: Receive = {
    case evt: Event =>
      eventBuffer = evt :: eventBuffer
      if (eventBuffer.size >= batchSize) {
        persistAll(eventBuffer.reverse) { _ =>
          eventBuffer = List.empty
          // 批量处理完成逻辑
        }
      }
    case FlushBatch =>
      if (eventBuffer.nonEmpty) {
        persistAll(eventBuffer.reverse) { _ =>
          eventBuffer = List.empty
        }
      }
  }
}

可配合定时任务，确保未达批量阈值的事件也能定时持久化，避免数据丢失。

4. 优化Actor生命周期与持久化策略 #

• 及时终止闲置Actor：对短生命周期或不再使用的Actor，调用context.stop(self)终止，释放连接池资源（连接会自动归还到池）。
• 启用快照机制：定期对Actor状态快照，减少恢复时查询的事件日志数量，降低数据库连接使用频次：

override def receiveRecover: Receive = {
  case SnapshotOffer(_, snapshot: MyState) =>
    currentState = snapshot
  case evt: Event =>
    updateState(evt)
}

// 在业务逻辑节点触发快照
private def takeSnapshot(): Unit = {
  saveSnapshot(currentState)
}

• 合并小事件：将多个关联小事件合并为一个大事件持久化，减少数据库交互次数。

内容的提问来源于stack exchange，提问作者Liron Levy