经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
restapi（8）- restapi-sql：用户自主的服务
来源：cnblogs　　作者：雪川大虫　　时间：2019/10/29 10:29:18　　对本文有异议

  学习函数式编程初衷是看到自己熟悉的oop编程语言和sql数据库在现代商业社会中前景暗淡，准备完全放弃windows技术栈转到分布式大数据技术领域的。但是在现实中理想总是不如人意，本来想在一个规模较小的公司展展拳脚，以为小公司会少点历史包袱，有利于全面技术改造。但现实是：即使是小公司，一旦有个成熟的产品，那么进行全面的技术更新基本上是不可能的了，因为公司要生存，开发人员很难新旧技术之间随时切换。除非有狂热的热情，员工怠慢甚至抵制情绪不容易解决。只能采取逐步切换方式：保留原有产品的后期维护不动，新产品开发用一些新的技术。在我们这里的情况就是：以前一堆c#、sqlserver的东西必须保留，新的功能比如大数据、ai、识别等必须用新的手段如scala、python、dart、akka、kafka、cassandra、mongodb来开发。好了，新旧两个开发平台之间的软件系统对接又变成了一个问题。

   现在我们这里有个需求：把在linux-ubuntu akka-cluster集群环境里mongodb里数据处理的结果传给windows server下SQLServer里。这是一种典型的异系统集成场景。我的解决方案是通过一个restapi服务作为两个系统的数据桥梁，这个restapi的最基本要求是：

1、支持任何操作系统前端：这个没什么问题，在http层上通过json交换数据

2、能读写mongodb：在前面讨论的restapi-mongo已经实现了这一功能

3、能读写windows server环境下的sqlserver：这个是本篇讨论的主题

4、用户能够比较方便的对平台数据库进行操作，最好免去前后双方每类操作都需要进行协定model这一过程，也就是能达到用户随意调用服务

前面曾经实现了一个jdbc-engine项目，基于scalikejdbc，不过只示范了slick-h2相关的功能。现在需要sqlserver-jdbc驱动，然后试试能不能在JVM里驱动windows下的sqlserver。maven里找不到sqlserver的驱动，但从微软官网可以下载mssql-jdbc-7.0.0.jre8.jar。这是个jar，在sbt里称作unmanagedjar，不能摆在build.sbt的dependency里。这个需要摆在项目根目录下的lib目录下即可（也可以在放在build.sbt里unmanagedBase :=?? 指定的路径下）。然后是数据库连接，下面是可以使用sqlserver的application.conf配置文件内容：

  1. # JDBC settings
  2. prod {
  3.   db {
  4.     h2 {
  5.       driver = "org.h2.Driver"
  6.       url = "jdbc:h2:tcp://localhost/~/slickdemo"
  7.       user = ""
  8.       password = ""
  9.       poolFactoryName = "hikaricp"
  10.       numThreads = 10
  11.       maxConnections = 12
  12.       minConnections = 4
  13.       keepAliveConnection = true
  14.     }
  15.     mysql {
  16.       driver = "com.mysql.cj.jdbc.Driver"
  17.       url = "jdbc:mysql://localhost:3306/testdb"
  18.       user = "root"
  19.       password = "123"
  20.       poolFactoryName = "hikaricp"
  21.       numThreads = 10
  22.       maxConnections = 12
  23.       minConnections = 4
  24.       keepAliveConnection = true
  25.     }
  26.     postgres {
  27.       driver = "org.postgresql.Driver"
  28.       url = "jdbc:postgresql://localhost:5432/testdb"
  29.       user = "root"
  30.       password = "123"
  31.       poolFactoryName = "hikaricp"
  32.       numThreads = 10
  33.       maxConnections = 12
  34.       minConnections = 4
  35.       keepAliveConnection = true
  36.     }
  37.     mssql {
  38.       driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
  39.       url = "jdbc:sqlserver://192.168.11.164:1433;integratedSecurity=false;Connect Timeout=3000"
  40.       user = "sa"
  41.       password = "Tiger2020"
  42.       poolFactoryName = "hikaricp"
  43.       numThreads = 10
  44.       maxConnections = 12
  45.       minConnections = 4
  46.       keepAliveConnection = true
  47.       connectionTimeout = 3000
  48.     }
  49.     termtxns {
  50.       driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
  51.       url = "jdbc:sqlserver://192.168.11.164:1433;DATABASE=TERMTXNS;integratedSecurity=false;Connect Timeout=3000"
  52.       user = "sa"
  53.       password = "Tiger2020"
  54.       poolFactoryName = "hikaricp"
  55.       numThreads = 10
  56.       maxConnections = 12
  57.       minConnections = 4
  58.       keepAliveConnection = true
  59.       connectionTimeout = 3000
  60.     }
  61.     crmdb {
  62.       driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
  63.       url = "jdbc:sqlserver://192.168.11.164:1433;DATABASE=CRMDB;integratedSecurity=false;Connect Timeout=3000"
  64.       user = "sa"
  65.       password = "Tiger2020"
  66.       poolFactoryName = "hikaricp"
  67.       numThreads = 10
  68.       maxConnections = 12
  69.       minConnections = 4
  70.       keepAliveConnection = true
  71.       connectionTimeout = 3000
  72.     }
  73.   }
  74.   # scallikejdbc Global settings
  75.   scalikejdbc.global.loggingSQLAndTime.enabled = true
  76.   scalikejdbc.global.loggingSQLAndTime.logLevel = info
  77.   scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  78.   scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  79.   scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  80.   scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  81.   scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  82.   scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
  83. }

这个文件里的mssql,termtxns,crmdb段落都是给sqlserver的，它们都使用hikaricp线程池管理。

在jdbc-engine里启动数据库方式如下：

  1.   ConfigDBsWithEnv("prod").setup('termtxns)
  2.   ConfigDBsWithEnv("prod").setup('crmdb)
  3.   ConfigDBsWithEnv("prod").loadGlobalSettings()

这段打开了在配置文件中用termtxns,crmdb注明的数据库。

下面是SqlHttpServer.scala的代码：

  1. package com.datatech.rest.sql
  2. import akka.http.scaladsl.Http
  3. import akka.http.scaladsl.server.Directives._
  4. import pdi.jwt._
  5. import AuthBase._
  6. import MockUserAuthService._
  7. import com.datatech.sdp.jdbc.config.ConfigDBsWithEnv
  9. import akka.actor.ActorSystem
  10. import akka.stream.ActorMaterializer
  12. import Repo._
  13. import SqlRoute._
  15. object SqlHttpServer extends App {
  17.   implicit val httpSys = ActorSystem("sql-http-sys")
  18.   implicit val httpMat = ActorMaterializer()
  19.   implicit val httpEC = httpSys.dispatcher
  21.   ConfigDBsWithEnv("prod").setup('termtxns)
  22.   ConfigDBsWithEnv("prod").setup('crmdb)
  23.   ConfigDBsWithEnv("prod").loadGlobalSettings()
  25.   implicit val authenticator = new AuthBase()
  26.     .withAlgorithm(JwtAlgorithm.HS256)
  27.     .withSecretKey("OpenSesame")
  28.     .withUserFunc(getValidUser)
  30.   val route =
  31.     path("auth") {
  32.       authenticateBasic(realm = "auth", authenticator.getUserInfo) { userinfo =>
  33.         post { complete(authenticator.issueJwt(userinfo))}
  34.       }
  35.     } ~
  36.       pathPrefix("api") {
  37.         authenticateOAuth2(realm = "api", authenticator.authenticateToken) { token =>
  38.           new SqlRoute("sql", token)(new JDBCRepo)
  39.             .route
  40.           // ~ ...
  41.         }
  42.       }
  44.   val (port, host) = (50081,"192.168.11.189")
  46.   val bindingFuture = Http().bindAndHandle(route,host,port)
  48.   println(s"Server running at $host $port. Press any key to exit ...")
  50.   scala.io.StdIn.readLine()
  51.   
  52.   bindingFuture.flatMap(_.unbind())
  53.     .onComplete(_ => httpSys.terminate())
  55. }

服务入口在http://mydemo.com/api/sql，服务包括get,post,put三类，参考这个SqlRoute：

  1. package com.datatech.rest.sql
  2. import akka.http.scaladsl.server.Directives
  3. import akka.stream.ActorMaterializer
  4. import akka.http.scaladsl.model._
  5. import akka.actor.ActorSystem
  6. import com.datatech.rest.sql.Repo.JDBCRepo
  7. import akka.http.scaladsl.common._
  8. import spray.json.DefaultJsonProtocol
  9. import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
  11. trait JsFormats extends SprayJsonSupport with DefaultJsonProtocol
  12. object JsConverters extends JsFormats {
  13.   import SqlModels._
  14.   implicit val brandFormat = jsonFormat2(Brand)
  15.   implicit val customerFormat = jsonFormat6(Customer)
  16. }
  18. object SqlRoute {
  19.   import JsConverters._
  20.   implicit val jsonStreamingSupport = EntityStreamingSupport.json()
  21.     .withParallelMarshalling(parallelism = 8, unordered = false)
  23.   class SqlRoute(val pathName: String, val jwt: String)(repo: JDBCRepo)(
  24.   implicit  sys: ActorSystem, mat: ActorMaterializer) extends Directives with JsonConverter {
  25.     val route = pathPrefix(pathName) {
  26.       path(Segment / Remaining) { case (db, tbl) =>
  27.         (get & parameter('sqltext)) { sql => {
  28.           val rsc = new RSConverter
  29.           val rows = repo.query[Map[String,Any]](db, sql, rsc.resultSet2Map)
  30.           complete(rows.map(m => toJson(m)))
  31.         }
  32.         } ~ (post & parameter('sqltext)) { sql =>
  33.               entity(as[String]){ json =>
  34.                 repo.batchInsert(db,tbl,sql,json)
  35.                 complete(StatusCodes.OK)
  36.               }
  37.         } ~ put {
  38.           entity(as[Seq[String]]) { sqls =>
  39.             repo.update(db, sqls)
  40.             complete(StatusCodes.OK)
  41.           }
  42.         }
  43.       }
  44.     }
  45.   }
  46. }

jdbc-engine的特点是可以用字符类型的sql语句来操作。所以我们可以通过传递字符串型的sql语句来实现服务调用，使用门槛低，方便通用。restapi-sql提供的是对服务器端sqlserver的普通操作，包括读get,写入post,更改put。这些sqlserver操作部分是在JDBCRepo里的：

  1. package com.datatech.rest.sql
  2. import com.datatech.sdp.jdbc.engine.JDBCEngine._
  3. import com.datatech.sdp.jdbc.engine.{JDBCQueryContext, JDBCUpdateContext}
  4. import scalikejdbc._
  5. import akka.stream.ActorMaterializer
  6. import com.datatech.sdp.result.DBOResult.DBOResult
  7. import akka.stream.scaladsl._
  8. import scala.concurrent._
  9. import SqlModels._
  11. object Repo {
  13.   class JDBCRepo(implicit ec: ExecutionContextExecutor, mat: ActorMaterializer) {
  14.     def query[R](db: String, sqlText: String, toRow: WrappedResultSet => R): Source[R,Any] = {
  15.       //construct the context
  16.       val ctx = JDBCQueryContext(
  17.         dbName = Symbol(db),
  18.         statement = sqlText
  19.       )
  20.       jdbcAkkaStream(ctx,toRow)
  21.     }
  23.     def query(db: String, tbl: String, sqlText: String) = {
  24.       //construct the context
  25.       val ctx = JDBCQueryContext(
  26.         dbName = Symbol(db),
  27.         statement = sqlText
  28.       )
  29.       jdbcQueryResult[Vector,RS](ctx,getConverter(tbl)).toFuture[Vector[RS]]
  30.     }
  32.     def update(db: String, sqlTexts: Seq[String]): DBOResult[Seq[Long]] = {
  33.       val ctx = JDBCUpdateContext(
  34.         dbName = Symbol(db),
  35.         statements = sqlTexts
  36.       )
  37.       jdbcTxUpdates(ctx)
  38.     }
  39.     def bulkInsert[P](db: String, sqlText: String, prepParams: P => Seq[Any], params: Source[P,_]) = {
  40.       val insertAction = JDBCActionStream(
  41.         dbName = Symbol(db),
  42.         parallelism = 4,
  43.         processInOrder = false,
  44.         statement = sqlText,
  45.         prepareParams = prepParams
  46.       )
  47.       params.via(insertAction.performOnRow).to(Sink.ignore).run()
  48.     }
  49.     def batchInsert(db: String, tbl: String, sqlText: String, jsonParams: String):DBOResult[Seq[Long]] = {
  50.       val ctx = JDBCUpdateContext(
  51.         dbName = Symbol(db),
  52.         statements = Seq(sqlText),
  53.         batch = true,
  54.         parameters = getSeqParams(jsonParams,sqlText)
  55.       )
  56.       jdbcBatchUpdate[Seq](ctx)
  57.     }
  58.   }
  59.   import monix.execution.Scheduler.Implicits.global
  60.   implicit class DBResultToFuture(dbr: DBOResult[_]){
  61.     def toFuture[R] = {
  62.       dbr.value.value.runToFuture.map {
  63.         eor =>
  64.           eor match {
  65.             case Right(or) => or match {
  66.               case Some(r) => r.asInstanceOf[R]
  67.               case None => throw new RuntimeException("Operation produced None result!")
  68.             }
  69.             case Left(err) => throw new RuntimeException(err)
  70.           }
  71.       }
  72.     }
  73.   }
  74. }

读query部分即 def query[R](db: String, sqlText: String, toRow: WrappedResultSet => R): Source[R,Any] = {...} 这个函数返回Source[R,Any]，下面我们好好谈谈这个R：R是读的结果，通常是某个类或model，比如读取Person记录返回一组Person类的实例。这里有一种强类型的感觉。一开始我也是随大流坚持建model后用toJson[E],fromJson[E]这样做线上数据转换。现在的问题是restapi-sql是一项公共服务，使用者知道sqlserver上有些什么表，然后希望通过sql语句来从这些表里读取数据。这些sql语句可能超出表的界限如sql join, union等，如果我们坚持每个返回结果都必须有个对应的model，那么显然就会牺牲这个服务的通用性。实际上，http线上数据交换本身就不可能是强类型的，因为经过了json转换。对于json转换来说，只要求字段名称、字段类型对称就行了。至于从什么类型转换成了另一个什么类型都没问题。所以，字段名+字段值的表现形式不就是Map[K,V]吗，我们就用Map[K,V]作为万能model就行了，没人知道。也就是说用户方通过sql语句指定返回的字段名称，它们可能是任何类型Any，具体类型自然会由数据库补上。服务方从数据库读取结果ResultSet后转成Map[K,V]然后再转成json返回给用户，用户可以用Map[String,Any]信息产生任何类型，这就是自主。好，就来看看如何将ResultSet转成Map[String,Any]：

  1. package com.datatech.rest.sql
  2. import scalikejdbc._
  3. import java.sql.ResultSetMetaData
  4. class RSConverter {
  5.   import RSConverterUtil._
  6.   var rsMeta: ResultSetMetaData = _
  7.   var columnCount: Int = 0
  8.   var rsFields: List[(String,String)] = List[(String,String)]()
  10.   def getFieldsInfo:List[(String,String)] =
  11.     ( 1 until columnCount).foldLeft(List[(String,String)]()) {
  12.     case (cons,i) =>
  13.       (rsMeta.getColumnLabel(i) -> rsMeta.getColumnTypeName(i)) :: cons
  14.   }
  15.   def resultSet2Map(rs: WrappedResultSet): Map[String,Any] = {
  16.     if(columnCount == 0) {
  17.       rsMeta =  rs.underlying.getMetaData
  18.       columnCount = rsMeta.getColumnCount
  19.       rsFields = getFieldsInfo
  20.     }
  21.     rsFields.foldLeft(Map[String,Any]()) {
  22.       case (m,(n,t)) =>
  23.         m + (n -> rsFieldValue(n,t,rs))
  24.     }
  25.   }
  26. }
  27. object RSConverterUtil {
  28.   import scala.collection.immutable.TreeMap
  29.   def map2Params(stm: String, m: Map[String,Any]): Seq[Any] = {
  30.     val sortedParams = m.foldLeft(TreeMap[Int,Any]()) {
  31.       case (t,(k,v)) => t + (stm.indexOfSlice(k) -> v)
  32.     }
  33.     sortedParams.map(_._2).toSeq
  34.   }
  35.   def rsFieldValue(fldname: String, fldType: String, rs: WrappedResultSet): Any = fldType match {
  36.     case "LONGVARCHAR" => rs.string(fldname)
  37.     case "VARCHAR" => rs.string(fldname)
  38.     case "CHAR" => rs.string(fldname)
  39.     case "BIT" => rs.boolean(fldname)
  40.     case "TIME" => rs.time(fldname)
  41.     case "TIMESTAMP" => rs.timestamp(fldname)
  42.     case "ARRAY" => rs.array(fldname)
  43.     case "NUMERIC" => rs.bigDecimal(fldname)
  44.     case "BLOB" => rs.blob(fldname)
  45.     case "TINYINT" => rs.byte(fldname)
  46.     case "VARBINARY" => rs.bytes(fldname)
  47.     case "BINARY" => rs.bytes(fldname)
  48.     case "CLOB" => rs.clob(fldname)
  49.     case "DATE" => rs.date(fldname)
  50.     case "DOUBLE" => rs.double(fldname)
  51.     case "REAL" => rs.float(fldname)
  52.     case "FLOAT" => rs.float(fldname)
  53.     case "INTEGER" => rs.int(fldname)
  54.     case "SMALLINT" => rs.int(fldname)
  55.     case "Option[Int]" => rs.intOpt(fldname)
  56.     case "BIGINT" => rs.long(fldname)
  57.   }
  58. }

这段主要功能是将JDBC的ResultSet转换成Map[String,Any]。在前面讨论的restapi-mongo我们可以进行Document到Map[String,Any]的转换以实现同样的目的。

下面是个调用query服务的例子：

  1.     val getAllRequest = HttpRequest(
  2.       HttpMethods.GET,
  3.       uri = "http://192.168.11.189:50081/api/sql/termtxns/brand?sqltext=SELECT%20*%20FROM%20BRAND",
  4.     ).addHeader(authentication)
  6.     (for {
  7.       response <- Http().singleRequest(getAllRequest)
  8.       json <- Unmarshal(response.entity).to[String]
  9.     } yield message).andThen {
  10.       case Success(msg) => println(s"Received json collection: $json")
  11.       case Failure(err) => println(s"Error: ${err.getMessage}")
  12.     }

特点是我只需要提供sql语句，服务就会返回一个json数组，然后我怎么把json转成任何类型就随我高兴了。

再看看post服务：在这里希望实现一种批次型插入表的功能，比如从一个数据表里把数据搬到另外一个表。一般来讲在jdbc操作里首先得提供一个模版，如：insert into person(fullname,code) values(?,?)，然后通过提供一组参数值来实现批次插入。当然，为安全起见，我们还是需要确定正确的参数位置，这个可以从sql语句里获取：

  1.   def map2Params(stm: String, m: Map[String,Any]): Seq[Any] = {
  2.     val sortedParams = m.foldLeft(TreeMap[Int,Any]()) {
  3.       case (t,(k,v)) => t + (stm.toUpperCase.indexOfSlice(k.toUpperCase) -> v)
  4.     }
  5.     sortedParams.map(_._2).toSeq
  6.   }
  8.   def getSeqParams(json: String, sql: String): Seq[Seq[Any]] = {
  9.     val seqOfjson = fromJson[Seq[String]](json)
  10.     val prs = seqOfjson.map(fromJson[Map[String,Any]])
  11.     prs.map(RSConverterUtil.map2Params(sql,_))
  12.   }

下面是个批次插入的示范代码：

  1.     val encodedSelect = URLEncode.encode("select id as code, name as fullname from members")
  2.     val encodedInsert = URLEncode.encode("insert into person(fullname,code) values(?,?)")
  3.     val getMembers = HttpRequest(
  4.        HttpMethods.GET,
  5.        uri = "http://192.168.0.189:50081/api/sql/h2/members?sqltext="+encodedSelect
  6.       ).addHeader(authentication)
  7.     val postRequest = HttpRequest(
  8.       HttpMethods.POST,
  9.       uri = "http://192.168.0.189:50081/api/sql/h2/person?sqltext="+encodedInsert,
  10.     ).addHeader(authentication)
  11.     
  12.     (for {
  13.       _ <- update("http://192.168.0.189:50081/api/sql/h2/person",Seq(createCTX))
  14.       respMembers <- Http().singleRequest(getMembers)
  15.       message <- Unmarshal(respMembers.entity).to[String]
  16.       reqEntity <- Marshal(message).to[RequestEntity]
  17.       respInsert <- Http().singleRequest(postRequest.copy(entity = reqEntity))
  18.  //       HttpEntity(ContentTypes.`application/json`,ByteString(message))))
  19.     } yield respInsert).onComplete {
  20.       case Success(r@HttpResponse(StatusCodes.OK, _, entity, _)) =>
  21.         println("builk insert successful!")
  22.       case Success(_) => println("builk insert failed!")
  23.       case Failure(err) => println(s"Error: ${err.getMessage}")
  24.     }

你看，我特别把参数值清单里字段位置和insert sql里字段先后位置颠倒了，但还是得到正确的结果。

最后是put：这是为批次型的事物处理设计的。接受一条或者多条无参数sql指令，多条指令会在一个事物中执行。具体使用方式如下：

  1.     def update(url: String, cmds: Seq[String])(implicit token: Authorization): Future[HttpResponse] =
  2.     for {
  3.       reqEntity <- Marshal(cmds).to[RequestEntity]
  4.       response <- Http().singleRequest(HttpRequest(
  5.         method=HttpMethods.PUT,uri=url,entity=reqEntity)
  6.       .addHeader(token))
  7.     } yield response

在上面的讨论里介绍了基于sqlserver的rest服务，与前面讨论的restapi-mongo从原理上区别并不大，重点是实现了用户主导的数据库操作。

 

原文链接：http://www.cnblogs.com/tiger-xc/p/11754487.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Scala热门文章

  • scala (3) Function 和 Method
  • flink 两个datastream实现left_join的两种方法
  • scala case class + BeanProperty+fastjson 快速实现类的序列化与反序列化
  • Scala解析MDF文件
  • flink 如何实现对watermark 的checkpoint,防止数据复写
  • Spark笔记(一):错误总结
  • 踩坑 scala 打包上传 找不到类
  • akka-typed(2) - typed-actor交流方式和交流协议
  • Scala(二)：元组、数组、映射
  • scala 命名规范

    • Scala推荐文章

  • Scala中Array和List的区别说明
  • alpakka-kafka(5)-kafka集群配置与分布式应用部署
  • alpakka-kafka(2)-consumer
  • alpakka-kafka(1)-producer
  • akka-typed(1) - actor生命周期管理
  • akka-typed(2) - typed-actor交流方式和交流协议
  • akka-typed(3) - PersistentActor has EventSourcedBehavior
  • akka-typed(4) - EventSourcedBehavior in action
  • kka-typed(5) - cluster：集群节点状态监视
  • akka-typed(6) - cluster：group router, cluster-load-balancing

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号