SDP(7):Cassandra- Cassandra-Engine:Streaming

小说:如何应对宝宝的第一个叛逆期!作者:安文纯更新时间:2019-03-25字数:61778

听到这个消息后,大家大都是感到高兴,所有人都不爽那个国务大臣了,现在他终于死了,可以说是举城欢庆啊。

【情感问答】异性恐惧症,不敢和男性谈恋爱

临走时,其中一个侦察兵兄弟朝韩非这边做了个手势,韩非立即领会,马上带着手下兄弟们迅速起身,朝祠堂两侧摸了上来!
不过断浪毕竟不是独孤求败,而步惊云也看出了断浪剑招的端倪所在,所以立刻改变招式,以封锁阻隔为主,事实上他这么做的确很好,一边的刘皓也对步惊云的做法露出了一丝赞赏,如果换了刘皓他有很多种方法解决,但是对于目前的断浪来说却只有一种,能否做到就看断浪的心能否和他手中的独孤九剑一般锐不可挡了。

唐三哼了一声,“要是你的无敌身次数已经用完了,你还不是一样会冲上来?”

  akka在alpakka工具包里提供了对cassandra数据库的streaming功能。简单来讲就是用一个CQL-statement读取cassandra数据并产生akka-stream的Source。这是一个支持reactive-stream协议的流:

object CassandraSource {

  /**
   * Scala API: creates a [[CassandraSourceStage]] from a given statement.
   */
  def apply(stmt: Statement)(implicit session: Session): Source[Row, NotUsed] =
    Source.fromGraph(new CassandraSourceStage(Future.successful(stmt), session))

  /**
   * Scala API: creates a [[CassandraSourceStage]] from the result of a given Future.
   */
  def fromFuture(futStmt: Future[Statement])(implicit session: Session): Source[Row, NotUsed] =
    Source.fromGraph(new CassandraSourceStage(futStmt, session))

}

CassandraSource.apply构建Source[Row,NotUsed]。可以直接接通Flow[Row,Row,NotUsed]和Sink来使用。我们是通过CQLQueryContext来构建这个Source的:

  def cassandraStream[A](ctx: CQLQueryContext[A])
       (implicit session: Session, ec: ExecutionContextExecutor): Source[A,NotUsed] = {

    val prepStmt = session.prepare(ctx.statement)
    var boundStmt =  prepStmt.bind()
    val params = processParameters(ctx.parameter)
    boundStmt = prepStmt.bind(params:_*)
    ctx.consistency.foreach {consistency =>
      boundStmt.setConsistencyLevel(consistencyLevel(consistency))}

    CassandraSource(boundStmt.setFetchSize(ctx.fetchSize)).map(ctx.extractor)
   }

CQLQueryContext[A]在上期介绍过: 

case class CQLQueryContext[M](
                               statement: String,
                               extractor: Row => M,
                               parameter: Seq[Object] = Nil,
                               consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None,
                               fetchSize: Int = 100
                             ) { ctx =>
  def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLQueryContext[M] =
     ctx.copy(consistency = Some(_consistency))
  def setFetchSize(pageSize: Int): CQLQueryContext[M] =
    ctx.copy(fetchSize = pageSize)
}
object CQLQueryContext {
  def apply[M](stmt: String, converter: Row => M): CQLQueryContext[M] =
    new CQLQueryContext[M](statement = stmt, extractor = converter)
}

下面是一个流的构建和使用示范:

  case class Model (
                       rowid: Long,
                       measureid: Long,
                       state: String,
                       county: String,
                       year: Int,
                       value: Int,
                       createdAt: java.util.Date
                     )
  //data row converter
  val toModel = (rs: Row) => Model(
    rowid = rs.getLong("ROWID"),
    measureid = rs.getLong("MEASUREID"),
    state = rs.getString("STATENAME"),
    county = rs.getString("COUNTYNAME"),
    year = rs.getInt("REPORTYEAR"),
    value = rs.getInt("VALUE"),
    createdAt = rs.getTimestamp("CREATED")
  )

  //setup context
  val ctx = CQLQueryContext("select * from testdb.aqmrpt",toModel)
  //construct source
  val src = cassandraStream(ctx)
  //a display sink
  val snk = Sink.foreach[Model]{ r =>
    println(s"${r.rowid} ${r.state} ${r.county} ${r.year} ${r.value}")
  }
  //run on source
  src.to(snk).run()

除了通过读取数据构成stream source之外,我们还可以以流元素为数据进行数据库更新操作,因为我们可以用map来运行execute:

  case class CassandraActionStream[R](parallelism: Int = 1, processInOrder: Boolean = true,
                                  statement: String, prepareParams: R => Seq[Object],
                                  consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None){ cas =>
    def setParallelism(parLevel: Int): CassandraActionStream[R] = cas.copy(parallelism=parLevel)
    def setProcessOrder(ordered: Boolean): CassandraActionStream[R] = cas.copy(processInOrder = ordered)
    def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CassandraActionStream[R] =
      cas.copy(consistency = Some(_consistency))

    def perform(r: R)(implicit session: Session, ec: ExecutionContext) = {
      val prepStmt = session.prepare(statement)
      var boundStmt =  prepStmt.bind()
      val params = processParameters(prepareParams(r))
      boundStmt = prepStmt.bind(params:_*)
      consistency.foreach { cons =>
        boundStmt.setConsistencyLevel(CQLContext.consistencyLevel(cons))
      }
      session.executeAsync(boundStmt).map(_ => r)
    }
    def performOnRow(implicit session: Session, ec: ExecutionContext): Flow[R, R, NotUsed] =
      if (processInOrder)
        Flow[R].mapAsync(parallelism)(perform)
      else
        Flow[R].mapAsyncUnordered(parallelism)(perform)

  }
  object CassandraActionStream {
    def apply[R](_statement: String, params: R => Seq[Object]): CassandraActionStream[R] =
      new CassandraActionStream[R]( statement=_statement, prepareParams = params)
  }

CassandraActionStream可以用statement,params构建。它的一个函数performOnRow是一个Flow[R,R,NotUsed],可以把每个R转换成一条CQL后通过map来运行executeAsyn,造成一种批次运算效果。下面是CassandraActionStream的使用示范:

  //pass context to construct akka-source
  val jdbcSource = jdbcAkkaStream(ctx)

  val cqlInsert = "insert into testdb.AQMRPT(ROWID,MEASUREID,STATENAME," +
    "COUNTYNAME,REPORTYEAR,VALUE,CREATED) VALUES(?,?,?,?,?,?,?)"

  val toPparams: DataRow => Seq[Object] = r => {
    Seq[Object](r.rowid.asInstanceOf[Object],
      r.measureid.asInstanceOf[Object],
      r.state,
      r.county,
      r.year.asInstanceOf[Object],
      r.value.asInstanceOf[Object],
      CQLDateTimeNow
    )
  }

  val actionStream = CassandraActionStream(cqlInsert,toPparams).setParallelism(2)
    .setProcessOrder(false)
  val actionFlow: Flow[DataRow,DataRow,NotUsed] = actionStream.performOnRow

  val sink = Sink.foreach[DataRow]{ r =>
    println(s"${r.rowid} ${r.state} ${r.county} ${r.year} ${r.value}")
  }
  val sts = jdbcSource.take(100).via(actionFlow).to(sink).run()

下面的例子里我们用CassandraStream的流元素更新h2数据库中的数据,调用了JDBCActionStream: 

  //data row converter
  val toModel = (rs: Row) => Model(
    rowid = rs.getLong("ROWID"),
    measureid = rs.getLong("MEASUREID"),
    state = rs.getString("STATENAME"),
    county = rs.getString("COUNTYNAME"),
    year = rs.getInt("REPORTYEAR"),
    value = rs.getInt("VALUE"),
    createdAt = rs.getTimestamp("CREATED")
  )

  //setup context
  val cqlCtx = CQLQueryContext("select * from testdb.aqmrpt",toModel)
  //construct source
  val src = cassandraStream(cqlCtx)
  //a display sink
  val snk = Sink.foreach[Model]{ r =>
    println(s"${r.rowid} ${r.state} ${r.county} ${r.year} ${r.value}")
  }
  //run on source
  src.to(snk).run()

  val params: Model => Seq[Any] = row => {
    Seq((row.value * 10), row.rowid) }

  val jdbcActionStream = JDBCActionStream("h2, "update AQMRPT set total = ? where rowid = ?",params)
      .setParallelism(2).setProcessOrder(false)
  val jdbcActionFlow = jdbcActionStream.performOnRow

  //update rows in h2 database from data in cassandra database
  src.via(jdbcActionFlow).to(snk).run()

下面是本次示范的源代码:

build.sbt

name := "learn_cassandra"

version := "0.1"

scalaVersion := "2.12.4"

libraryDependencies := Seq(
  "com.datastax.cassandra" % "cassandra-driver-core" % "3.4.0",
  "com.datastax.cassandra" % "cassandra-driver-extras" % "3.4.0",
  "com.typesafe.akka" %% "akka-actor" % "2.5.4",
  "com.typesafe.akka" %% "akka-stream" % "2.5.4",
  "com.lightbend.akka" %% "akka-stream-alpakka-cassandra" % "0.16",
  "org.scalikejdbc" %% "scalikejdbc"       % "3.2.1",
  "org.scalikejdbc" %% "scalikejdbc-test"   % "3.2.1"   % "test",
  "org.scalikejdbc" %% "scalikejdbc-config"  % "3.2.1",
  "org.scalikejdbc" %% "scalikejdbc-streams" % "3.2.1",
  "org.scalikejdbc" %% "scalikejdbc-joda-time" % "3.2.1",
  "com.h2database"  %  "h2"                % "1.4.196",
  "mysql" % "mysql-connector-java" % "6.0.6",
  "org.postgresql" % "postgresql" % "42.2.0",
  "commons-dbcp" % "commons-dbcp" % "1.4",
  "org.apache.tomcat" % "tomcat-jdbc" % "9.0.2",
  "com.zaxxer" % "HikariCP" % "2.7.4",
  "com.jolbox" % "bonecp" % "0.8.0.RELEASE",
  "com.typesafe.slick" %% "slick" % "3.2.1",
  "ch.qos.logback"  %  "logback-classic"   % "1.2.3")

resources/application.conf

# JDBC settings
test {
  db {
    h2 {
      driver = "org.h2.Driver"
      url = "jdbc:h2:tcp://localhost/~/slickdemo"
      user = ""
      password = ""
      poolInitialSize = 5
      poolMaxSize = 7
      poolConnectionTimeoutMillis = 1000
      poolValidationQuery = "select 1 as one"
      poolFactoryName = "commons-dbcp2"
    }
  }

  db.mysql.driver = "com.mysql.cj.jdbc.Driver"
  db.mysql.url = "jdbc:mysql://localhost:3306/testdb"
  db.mysql.user = "root"
  db.mysql.password = "123"
  db.mysql.poolInitialSize = 5
  db.mysql.poolMaxSize = 7
  db.mysql.poolConnectionTimeoutMillis = 1000
  db.mysql.poolValidationQuery = "select 1 as one"
  db.mysql.poolFactoryName = "bonecp"

  # scallikejdbc Global settings
  scalikejdbc.global.loggingSQLAndTime.enabled = true
  scalikejdbc.global.loggingSQLAndTime.logLevel = info
  scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
}
dev {
  db {
    h2 {
      driver = "org.h2.Driver"
      url = "jdbc:h2:tcp://localhost/~/slickdemo"
      user = ""
      password = ""
      poolFactoryName = "hikaricp"
      numThreads = 10
      maxConnections = 12
      minConnections = 4
      keepAliveConnection = true
    }
    mysql {
      driver = "com.mysql.cj.jdbc.Driver"
      url = "jdbc:mysql://localhost:3306/testdb"
      user = "root"
      password = "123"
      poolInitialSize = 5
      poolMaxSize = 7
      poolConnectionTimeoutMillis = 1000
      poolValidationQuery = "select 1 as one"
      poolFactoryName = "bonecp"

    }
    postgres {
      driver = "org.postgresql.Driver"
      url = "jdbc:postgresql://localhost:5432/testdb"
      user = "root"
      password = "123"
      poolFactoryName = "hikaricp"
      numThreads = 10
      maxConnections = 12
      minConnections = 4
      keepAliveConnection = true
    }
  }
  # scallikejdbc Global settings
  scalikejdbc.global.loggingSQLAndTime.enabled = true
  scalikejdbc.global.loggingSQLAndTime.logLevel = info
  scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
}

CassandraEngine.scala

import com.datastax.driver.core._

import scala.concurrent._
import com.google.common.util.concurrent.{FutureCallback, Futures, ListenableFuture}

import scala.collection.JavaConverters._
import scala.collection.generic.CanBuildFrom
import scala.concurrent.duration.Duration
import akka.NotUsed
import akka.stream.alpakka.cassandra.scaladsl._
import akka.stream.scaladsl._

object CQLContext {
  // Consistency Levels
  type CONSISTENCY_LEVEL = Int
  val ANY: CONSISTENCY_LEVEL          =                                        0x0000
  val ONE: CONSISTENCY_LEVEL          =                                        0x0001
  val TWO: CONSISTENCY_LEVEL          =                                        0x0002
  val THREE: CONSISTENCY_LEVEL        =                                        0x0003
  val QUORUM : CONSISTENCY_LEVEL      =                                        0x0004
  val ALL: CONSISTENCY_LEVEL          =                                        0x0005
  val LOCAL_QUORUM: CONSISTENCY_LEVEL =                                        0x0006
  val EACH_QUORUM: CONSISTENCY_LEVEL  =                                        0x0007
  val LOCAL_ONE: CONSISTENCY_LEVEL    =                                      0x000A
  val LOCAL_SERIAL: CONSISTENCY_LEVEL =                                     0x000B
  val SERIAL: CONSISTENCY_LEVEL       =                                      0x000C

  def apply(): CQLContext = CQLContext(statements = Nil)

  def consistencyLevel: CONSISTENCY_LEVEL => ConsistencyLevel = consistency => {
    consistency match {
      case ALL => ConsistencyLevel.ALL
      case ONE => ConsistencyLevel.ONE
      case TWO => ConsistencyLevel.TWO
      case THREE => ConsistencyLevel.THREE
      case ANY => ConsistencyLevel.ANY
      case EACH_QUORUM => ConsistencyLevel.EACH_QUORUM
      case LOCAL_ONE => ConsistencyLevel.LOCAL_ONE
      case QUORUM => ConsistencyLevel.QUORUM
      case SERIAL => ConsistencyLevel.SERIAL
      case LOCAL_SERIAL => ConsistencyLevel.LOCAL_SERIAL

    }
  }

}
case class CQLQueryContext[M](
                               statement: String,
                               extractor: Row => M,
                               parameter: Seq[Object] = Nil,
                               consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None,
                               fetchSize: Int = 100
                             ) { ctx =>
  def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLQueryContext[M] =
     ctx.copy(consistency = Some(_consistency))
  def setFetchSize(pageSize: Int): CQLQueryContext[M] =
    ctx.copy(fetchSize = pageSize)
}
object CQLQueryContext {
  def apply[M](stmt: String, converter: Row => M): CQLQueryContext[M] =
    new CQLQueryContext[M](statement = stmt, extractor = converter)
}

case class CQLContext(
                       statements: Seq[String],
                       parameters: Seq[Seq[Object]] = Nil,
                       consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None
                     ) { ctx =>

  def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CQLContext =
    ctx.copy(consistency = Some(_consistency))
  def setCommand(_statement: String, _parameters: Object*): CQLContext =
    ctx.copy(statements = Seq(_statement), parameters = Seq(_parameters))
  def appendCommand(_statement: String, _parameters: Object*): CQLContext =
    ctx.copy(statements = ctx.statements :+ _statement,
      parameters = ctx.parameters ++ Seq(_parameters))
}

object CQLEngine {
  import CQLContext._
  import CQLHelpers._

  def fetchResultPage[C[_] <: TraversableOnce[_],A](ctx: CQLQueryContext[A], pageSize: Int = 100)(
    implicit session: Session, cbf: CanBuildFrom[Nothing, A, C[A]]): (ResultSet, C[A])= {

    val prepStmt = session.prepare(ctx.statement)

    var boundStmt =  prepStmt.bind()
    if (ctx.parameter != Nil) {
      val params = processParameters(ctx.parameter)
      boundStmt = prepStmt.bind(params:_*)
    }

    ctx.consistency.foreach {consistency =>
      boundStmt.setConsistencyLevel(consistencyLevel(consistency))}

    val resultSet = session.execute(boundStmt.setFetchSize(pageSize))
    (resultSet,(resultSet.asScala.view.map(ctx.extractor)).to[C])
  }
  def fetchMorePages[C[_] <: TraversableOnce[_],A](resultSet: ResultSet, timeOut: Duration)(
    extractor: Row => A)(implicit cbf: CanBuildFrom[Nothing, A, C[A]]): (ResultSet,Option[C[A]]) =
    if (resultSet.isFullyFetched) {
      (resultSet, None)
    } else {
      try {
        val result = Await.result(resultSet.fetchMoreResults(), timeOut)
        (result, Some((result.asScala.view.map(extractor)).to[C]))
      } catch { case e: Throwable => (resultSet, None) }
    }
  def cqlExecute(ctx: CQLContext)(
    implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
    if (ctx.statements.size == 1)
      cqlSingleUpdate(ctx)
    else
      cqlMultiUpdate(ctx)
  }
  def cqlSingleUpdate(ctx: CQLContext)(
    implicit session: Session, ec: ExecutionContext): Future[Boolean] = {

    val prepStmt = session.prepare(ctx.statements.head)

    var boundStmt =  prepStmt.bind()
    if (ctx.parameters != Nil) {
      val params = processParameters(ctx.parameters.head)
      boundStmt = prepStmt.bind(params:_*)
    }

    ctx.consistency.foreach {consistency =>
      boundStmt.setConsistencyLevel(consistencyLevel(consistency))}
    session.executeAsync(boundStmt).map(_.wasApplied())
  }
  def cqlMultiUpdate(ctx: CQLContext)(
    implicit session: Session, ec: ExecutionContext): Future[Boolean] = {
    val commands: Seq[(String,Seq[Object])] = ctx.statements zip ctx.parameters
    var batch = new BatchStatement()
    commands.foreach { case (stm, params) =>
      val prepStmt = session.prepare(stm)
      if (params == Nil)
        batch.add(prepStmt.bind())
      else {
        val p = processParameters(params)
        batch.add(prepStmt.bind(p: _*))
      }
    }
    ctx.consistency.foreach {consistency =>
      batch.setConsistencyLevel(consistencyLevel(consistency))}
    session.executeAsync(batch).map(_.wasApplied())
  }

  def cassandraStream[A](ctx: CQLQueryContext[A])
       (implicit session: Session, ec: ExecutionContextExecutor): Source[A,NotUsed] = {

    val prepStmt = session.prepare(ctx.statement)
    var boundStmt =  prepStmt.bind()
    val params = processParameters(ctx.parameter)
    boundStmt = prepStmt.bind(params:_*)
    ctx.consistency.foreach {consistency =>
      boundStmt.setConsistencyLevel(consistencyLevel(consistency))}

    CassandraSource(boundStmt.setFetchSize(ctx.fetchSize)).map(ctx.extractor)
   }

  case class CassandraActionStream[R](parallelism: Int = 1, processInOrder: Boolean = true,
                                  statement: String, prepareParams: R => Seq[Object],
                                  consistency: Option[CQLContext.CONSISTENCY_LEVEL] = None){ cas =>
    def setParallelism(parLevel: Int): CassandraActionStream[R] = cas.copy(parallelism=parLevel)
    def setProcessOrder(ordered: Boolean): CassandraActionStream[R] = cas.copy(processInOrder = ordered)
    def setConsistencyLevel(_consistency: CQLContext.CONSISTENCY_LEVEL): CassandraActionStream[R] =
      cas.copy(consistency = Some(_consistency))

    private def perform(r: R)(implicit session: Session, ec: ExecutionContext) = {
      val prepStmt = session.prepare(statement)
      var boundStmt =  prepStmt.bind()
      val params = processParameters(prepareParams(r))
      boundStmt = prepStmt.bind(params:_*)
      consistency.foreach { cons =>
        boundStmt.setConsistencyLevel(CQLContext.consistencyLevel(cons))
      }
      session.executeAsync(boundStmt).map(_ => r)
    }
    def performOnRow(implicit session: Session, ec: ExecutionContext): Flow[R, R, NotUsed] =
      if (processInOrder)
        Flow[R].mapAsync(parallelism)(perform)
      else
        Flow[R].mapAsyncUnordered(parallelism)(perform)

  }
  object CassandraActionStream {
    def apply[R](_statement: String, params: R => Seq[Object]): CassandraActionStream[R] =
      new CassandraActionStream[R]( statement=_statement, prepareParams = params)
  }

}
object CQLHelpers {
  import java.nio.ByteBuffer
  import java.io._
  import java.nio.file._
  import com.datastax.driver.core.LocalDate
  import com.datastax.driver.extras.codecs.jdk8.InstantCodec
  import java.time.Instant
  import akka.stream.scaladsl._
  import akka.stream._


  implicit def listenableFutureToFuture[T](
                                            listenableFuture: ListenableFuture[T]): Future[T] = {
    val promise = Promise[T]()
    Futures.addCallback(listenableFuture, new FutureCallback[T] {
      def onFailure(error: Throwable): Unit = {
        promise.failure(error)
        ()
      }
      def onSuccess(result: T): Unit = {
        promise.success(result)
        ()
      }
    })
    promise.future
  }

  case class CQLDate(year: Int, month: Int, day: Int)
  case object CQLTodayDate
  case class CQLDateTime(year: Int, Month: Int,
                         day: Int, hour: Int, minute: Int, second: Int, millisec: Int = 0)
  case object CQLDateTimeNow

  def processParameters(params: Seq[Object]): Seq[Object] = {
    import java.time.{Clock,ZoneId}
    params.map { obj =>
      obj match {
        case CQLDate(yy, mm, dd) => LocalDate.fromYearMonthDay(yy, mm, dd)
        case CQLTodayDate =>
          val today = java.time.LocalDate.now()
          LocalDate.fromYearMonthDay(today.getYear, today.getMonth.getValue, today.getDayOfMonth)
        case CQLDateTimeNow => Instant.now(Clock.system(ZoneId.of("EST", ZoneId.SHORT_IDS)))
        case CQLDateTime(yy, mm, dd, hr, ms, sc, mi) =>
          Instant.parse(f"$yy%4d-$mm%2d-$dd%2dT$hr%2d:$ms%2d:$sc%2d$mi%3d")
        case p@_ => p
      }
    }
  }
  class ByteBufferInputStream(buf: ByteBuffer) extends InputStream {
    override def read: Int = {
      if (!buf.hasRemaining) return -1
      buf.get
    }

    override def read(bytes: Array[Byte], off: Int, len: Int): Int = {
      val length: Int = Math.min(len, buf.remaining)
      buf.get(bytes, off, length)
      length
    }
  }
  object ByteBufferInputStream {
    def apply(buf: ByteBuffer): ByteBufferInputStream = {
      new ByteBufferInputStream(buf)
    }
  }
  class FixsizedByteBufferOutputStream(buf: ByteBuffer) extends OutputStream {

    override def write(b: Int): Unit = {
      buf.put(b.toByte)
    }

    override def write(bytes: Array[Byte], off: Int, len: Int): Unit = {
      buf.put(bytes, off, len)
    }
  }
  object FixsizedByteBufferOutputStream {
    def apply(buf: ByteBuffer) = new FixsizedByteBufferOutputStream(buf)
  }
  class ExpandingByteBufferOutputStream(var buf: ByteBuffer, onHeap: Boolean) extends OutputStream {

    private val increasing = ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR

    override def write(b: Array[Byte], off: Int, len: Int): Unit = {
      val position = buf.position
      val limit = buf.limit
      val newTotal: Long = position + len
      if(newTotal > limit){
        var capacity = (buf.capacity * increasing)
        while(capacity <= newTotal){
          capacity = (capacity*increasing)
        }
        increase(capacity.toInt)
      }

      buf.put(b, 0, len)
    }

    override def write(b: Int): Unit= {
      if (!buf.hasRemaining) increase((buf.capacity * increasing).toInt)
      buf.put(b.toByte)
    }
    protected def increase(newCapacity: Int): Unit = {
      buf.limit(buf.position)
      buf.rewind
      val newBuffer =
        if (onHeap) ByteBuffer.allocate(newCapacity)
        else  ByteBuffer.allocateDirect(newCapacity)
      newBuffer.put(buf)
      buf.clear
      buf = newBuffer
    }
    def size: Long = buf.position
    def capacity: Long = buf.capacity
    def byteBuffer: ByteBuffer = buf
  }
  object ExpandingByteBufferOutputStream {
    val DEFAULT_INCREASING_FACTOR = 1.5f
    def apply(size: Int, increasingBy: Float, onHeap: Boolean) = {
      if (increasingBy <= 1) throw new IllegalArgumentException("Increasing Factor must be greater than 1.0")
      val buffer: ByteBuffer =
        if (onHeap) ByteBuffer.allocate(size)
        else ByteBuffer.allocateDirect(size)
      new ExpandingByteBufferOutputStream(buffer,onHeap)
    }
    def apply(size: Int): ExpandingByteBufferOutputStream = {
      apply(size, ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR, false)
    }

    def apply(size: Int, onHeap: Boolean): ExpandingByteBufferOutputStream = {
      apply(size, ExpandingByteBufferOutputStream.DEFAULT_INCREASING_FACTOR, onHeap)
    }

    def apply(size: Int, increasingBy: Float): ExpandingByteBufferOutputStream = {
      apply(size, increasingBy, false)
    }

  }
  def cqlFileToBytes(fileName: String): ByteBuffer = {
    val fis = new FileInputStream(fileName)
    val b = new Array[Byte](fis.available + 1)
    val length = b.length
    fis.read(b)
    ByteBuffer.wrap(b)
  }
  def cqlBytesToFile(bytes: ByteBuffer, fileName: String)(
    implicit mat: Materializer): Future[IOResult] = {
    val source = StreamConverters.fromInputStream(() => ByteBufferInputStream(bytes))
    source.runWith(FileIO.toPath(Paths.get(fileName)))
  }
  def cqlDateTimeString(date: java.util.Date, fmt: String): String = {
    val outputFormat = new java.text.SimpleDateFormat(fmt)
    outputFormat.format(date)
  }
  def useJava8DateTime(cluster: Cluster) = {
    //for jdk8 datetime format
    cluster.getConfiguration().getCodecRegistry()
      .register(InstantCodec.instance)
  }
}

JDBCEngine.scala

package jdbccontext
import java.sql.PreparedStatement

import scala.collection.generic.CanBuildFrom
import akka.stream.scaladsl._
import scalikejdbc._
import scalikejdbc.streams._
import akka.NotUsed

import scala.util._
import scalikejdbc.TxBoundary.Try._

import scala.concurrent.ExecutionContextExecutor

object JDBCContext {
  type SQLTYPE = Int
  val SQL_EXEDDL= 1
  val SQL_UPDATE = 2
  val RETURN_GENERATED_KEYVALUE = true
  val RETURN_UPDATED_COUNT = false

}

case class JDBCQueryContext[M](
                                dbName: Symbol,
                                statement: String,
                                parameters: Seq[Any] = Nil,
                                fetchSize: Int = 100,
                                autoCommit: Boolean = false,
                                queryTimeout: Option[Int] = None,
                                extractor: WrappedResultSet => M)


case class JDBCContext(
                        dbName: Symbol,
                        statements: Seq[String] = Nil,
                        parameters: Seq[Seq[Any]] = Nil,
                        fetchSize: Int = 100,
                        queryTimeout: Option[Int] = None,
                        queryTags: Seq[String] = Nil,
                        sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_UPDATE,
                        batch: Boolean = false,
                        returnGeneratedKey: Seq[Option[Any]] = Nil,
                        // no return: None, return by index: Some(1), by name: Some("id")
                        preAction: Option[PreparedStatement => Unit] = None,
                        postAction: Option[PreparedStatement => Unit] = None) {

  ctx =>

  //helper functions

  def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag)

  def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags)

  def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size)

  def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time)

  def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
    if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
      !ctx.batch && ctx.statements.size == 1)
      ctx.copy(preAction = action)
    else
      throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  }

  def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
    if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
      !ctx.batch && ctx.statem

编辑:石丁

发布:2019-03-25 03:45:39

当前文章:http://leetaemin.cn/48619.html

教你欢乐相亲的3条黄金策略,纯干货 我们对校园凶杀案反思的重大失误导致凶杀案连续发生! 别和自己过不去 如何辨别公立医院整形科室是否外包 谢谢你的好意,可我不需要 真正的教育就是拼爹 经典禅诗:《灵魂》系列 圣诞节预热:诺心lecake星愿蛋糕真好吃!

57963 28598 54173 36665 38497 10026 55548 19841 98336 44660 41613 47389 35552 19720 84442 36934 36448 89838 91191 71711

我要说两句: (0人参与)

发布