Akka Slick替代Anorm的SQL(...).on(params)实现方案咨询

Got it, let's break down how to replicate your Anorm dynamic parameter insert logic with Akka Slick—especially since you need that flexible, dynamic parameter list instead of the simplified interpolated syntax.

First, a quick heads-up: Akka Slick operates asynchronously, so all database actions return Futures, which is a key shift from Anorm's synchronous execution. Let's cover both fixed-column dynamic inserts and fully dynamic (column + parameter) scenarios, plus how to tie this into Akka Streams.

1. Basic Table Mapping Setup #

First, define a table class to map your person table to Scala types—this is standard Slick boilerplate:

import slick.jdbc.MySQLProfile.api._ // Swap this for your DB profile (Postgres, H2, etc.)

class Persons(tag: Tag) extends Table[(String, Int)](tag, "person") {
  def name = column[String]("name")
  def age = column[Int]("age")
  def * = (name, age) // Projection for selecting/inserting full rows
}

val persons = TableQuery[Persons] // Query object for interacting with the table
val db = Database.forConfig("your-db-config") // Initialize your database connection

2. Dynamic Parameter Insert (Fixed Columns) #

If your columns are fixed (like name and age) but you need to assemble parameters dynamically (mirroring your Anorm Seq[NamedParameter]), you can map your dynamic parameters to a tuple and use Slick's += operator:

// Simulate dynamic parameter assembly (e.g., from user input, API payloads)
val dynamicParams = Map("name" -> "john", "age" -> 30)

// Convert params to the tuple type defined in your table projection
val personTuple = (dynamicParams("name"), dynamicParams("age").asInstanceOf[Int])

// Create and execute the insert action
val insertAction = persons += personTuple
db.run(insertAction).onComplete {
  case scala.util.Success(count) => println(s"Inserted $count row(s)")
  case scala.util.Failure(err) => println(s"Insert failed: ${err.getMessage}")
}

3. Fully Dynamic Insert (Dynamic Columns + Parameters) #

If you need maximum flexibility (e.g., columns are determined at runtime), use Slick's dynamic SQL support with sqlu (which returns an update count, just like Anorm's executeUpdate()):

// Dynamic columns and parameters (could come from any runtime source)
val dynamicColumns = List("name", "age")
val dynamicValues = List("john", 30)

// Build the dynamic INSERT statement
val insertSql = sqlu"""INSERT INTO person (#${dynamicColumns.mkString(", ")}) VALUES (#${dynamicValues.map(_ => "?").mkString(", ")})"""

// Bind the dynamic parameters—this is the direct equivalent of Anorm's `.on(params:_*)`
val boundAction = insertSql.bind(dynamicValues:_*)
db.run(boundAction)

4. Akka Streams Integration #

Since you're using Akka Streams, here's how to wire dynamic inserts into a stream for processing continuous data:

import akka.stream.scaladsl.{Sink, Source}
import scala.concurrent.ExecutionContext.Implicits.global

// Example source of dynamic person data (could be from Kafka, files, etc.)
val personSource = Source(List(
  Map("name" -> "john", "age" -> 30),
  Map("name" -> "jane", "age" -> 25),
  Map("name" -> "bob", "age" -> 35)
))

// Sink to process each dynamic record and insert asynchronously
val insertSink = Sink.foreachAsync(parallelism = 4) { params =>
  val personTuple = (params("name"), params("age").asInstanceOf[Int])
  db.run(persons += personTuple)
}

// Run the stream
personSource.runWith(insertSink)

For better performance with large streams, use batch inserts to reduce database round-trips:

val batchInsertSink = Sink.foldAsync(List.empty[(String, Int)]) { (acc, params) =>
  val personTuple = (params("name"), params("age").asInstanceOf[Int])
  val newAcc = acc :+ personTuple
  
  // Insert in batches of 100
  if (newAcc.size >= 100) {
    db.run(persons ++= newAcc).map(_ => List.empty)
  } else {
    Future.successful(newAcc)
  }
}.mapMaterializedValue { fut =>
  // Insert any remaining records after the stream ends
  fut.flatMap(remaining => db.run(persons ++= remaining))
}

personSource.runWith(batchInsertSink)

内容的提问来源于stack exchange，提问作者David Portabella