Akka(24): Stream:从外部系统控制数据流-control live stream from external system

 在数据流应用的现实场景中常常会遇到与外界系统对接的需求。这些外部系统可能是Actor系统又或者是一些其它类型的系统。与这些外界系统对接的意思是在另一个线程中运行的数据流可以接收外部系统推送的事件及做出行为改变的响应。

如果一个外界系统需要控制一个运行中数据流的功能环节GraphStage,首先必须在这个GraphStage内部构建一个控制函数,这样才能接触并更改GraphStage的内部状态。外部系统可以通过调用这个控制函数来向GraphStage发送信息,控制GraphStage行为。akka-stream是多线程异步模式的程序,所以这个函数只能是一个异步运行的回调callback。akka-stream提供了一个函数getAsyncCallback函数,能够把一个函数放到另一个线程里并返回它的callback:

  /**
   * Obtain a callback object that can be used asynchronously to re-enter the
   * current [[GraphStage]] with an asynchronous notification. The [[invoke()]] method of the returned
   * [[AsyncCallback]] is safe to be called from other threads and it will in the background thread-safely
   * delegate to the passed callback function. I.e. [[invoke()]] will be called by the external world and
   * the passed handler will be invoked eventually in a thread-safe way by the execution environment.
   *
   * This object can be cached and reused within the same [[GraphStageLogic]].
   */
  final def getAsyncCallback[T](handler: T ⇒ Unit): AsyncCallback[T] = {
    new AsyncCallback[T] {
      override def invoke(event: T): Unit =
        interpreter.onAsyncInput(GraphStageLogic.this, event, handler.asInstanceOf[Any ⇒ Unit])
    }
  }

getAsyncCallback把一个函数T=>Unit变成了异步运算函数并通过AsyncCallback返回它的回调callback。下面是getAsyncCallback的一个用例: 

//external system
object Injector {
  var callback: AsyncCallback[String] = null
   def inject(m: String) = {
     if (callback != null)
     callback.invoke(m)
   }
}
class InjectControl(injector: Injector.type) extends GraphStage[FlowShape[String,String]] {
  val inport = Inlet[String]("input")
  val outport = Outlet[String]("output")
  val shape = FlowShape.of(inport,outport)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) {
      var extMessage = ""
      override def preStart(): Unit = {
        val callback = getAsyncCallback[String] { m =>
          if (m.length > 0)
             m match {
               case "Stop" => completeStage()
               case s: String => extMessage = s
             }

        }
        injector.callback = callback
      }

      setHandler(inport, new InHandler {
        override def onPush(): Unit =
          if (extMessage.length > 0) {
            push(outport,extMessage)
            extMessage=""
          }
        else
          push(outport, grab(inport))
      })
      setHandler(outport, new OutHandler {
        override def onPull(): Unit = pull(inport)
      })
    }

}

上面这个例子里的object Injector模拟一个外部系统。我们重写了GraphStage InjectControl.createLogic里的preStart()函数,在这里把一个String=>Unit函数的callback登记在Injector里。这个callback函数能接受传入的String并更新内部状态extMessage,或者当传入String==“Stop"时终止数据流。在onPush()里extMessage最终会被当作流元素插入到数据流中。下面我们就构建这个GraphStage的测试运行程序:

object InteractWithStreams extends App {
  implicit val sys = ActorSystem("demoSys")
  implicit val ec = sys.dispatcher
  implicit val mat = ActorMaterializer(
    ActorMaterializerSettings(sys)
      .withInputBuffer(initialSize = 16, maxSize = 16)
  )

  val source = Source(Stream.from(1)).map(_.toString).delay(1.second,DelayOverflowStrategy.backpressure)
  val graph = new InjectControl(Injector)
  val flow = Flow.fromGraph(graph)

  source.via(flow).to(Sink.foreach(println)).run()
  Thread.sleep(2000)
  Injector.inject("hello")
  Thread.sleep(2000)
  Injector.inject("world!")
  Thread.sleep(2000)
  Injector.inject("Stop")

  scala.io.StdIn.readLine()
  sys.terminate()

}

试运行结果显示:

1
2
hello
4
5
6
world!
8
9
10


Process finished with exit code 0

正确地把"hello world!"插入了一个正在运行中的数据流中并在最后终止了这个数据流。

另外,一个GraphStage也可以被外界当作一种Actor来进行交流。我们可以在GraphStage内部构建一个(ActorRef,Any)=>Unit款式的函数,然后用getStageActor(func).ref以一种ActorRef形式返回这个函数:

 /**
   * Initialize a [[StageActorRef]] which can be used to interact with from the outside world "as-if" an [[Actor]].
   * The messages are looped through the [[getAsyncCallback]] mechanism of [[GraphStage]] so they are safe to modify
   * internal state of this stage.
   *
   * This method must (the earliest) be called after the [[GraphStageLogic]] constructor has finished running,
   * for example from the [[preStart]] callback the graph stage logic provides.
   *
   * Created [[StageActorRef]] to get messages and watch other actors in synchronous way.
   *
   * The [[StageActorRef]]'s lifecycle is bound to the Stage, in other words when the Stage is finished,
   * the Actor will be terminated as well. The entity backing the [[StageActorRef]] is not a real Actor,
   * but the [[GraphStageLogic]] itself, therefore it does not react to [[PoisonPill]].
   *
   * @param receive callback that will be called upon receiving of a message by this special Actor
   * @return minimal actor with watch method
   */
  // FIXME: I don't like the Pair allocation :(
  @ApiMayChange
  final protected def getStageActor(receive: ((ActorRef, Any)) ⇒ Unit): StageActor = {
    _stageActor match {
      case null ⇒
        val actorMaterializer = ActorMaterializerHelper.downcast(interpreter.materializer)
        _stageActor = new StageActor(actorMaterializer, getAsyncCallback, receive)
        _stageActor
      case existing ⇒
        existing.become(receive)
        existing
    }
  }
...
    /**
     * The ActorRef by which this StageActor can be contacted from the outside.
     * This is a full-fledged ActorRef that supports watching and being watched
     * as well as location transparent (remote) communication.
     */
    def ref: ActorRef = functionRef

下面是receive:((ActorRef,Any))=>Unit这个函数的实现例子:

      def behavior(m:(ActorRef,Any)): Unit = {
        val (sender, msg) = m

        msg.asInstanceOf[String] match {
          case "Stop" => completeStage()
          case s@ _ => extMessage = s
        }
      }

这个函数的输入参数(sender,msg)代表发送消息的Actor和发送的消息。与上个例子一样,作为一个GraphStage的内部函数,它可以使用、更新GraphStage内部状态。GraphStage的实现如下:

class StageAsActor(extActor: ActorRef) extends GraphStage[FlowShape[String,String]] {
  val inport = Inlet[String]("input")
  val outport = Outlet[String]("output")
  val shape = FlowShape.of(inport,outport)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) {
      var extMessage = ""
      override def preStart(): Unit = {
        extActor ! getStageActor(behavior).ref
      }

      def behavior(m:(ActorRef,Any)): Unit = {
        val (sender, msg) = m

        msg.asInstanceOf[String] match {
          case "Stop" => completeStage()
          case s@ _ => extMessage = s
        }
      }

      setHandler(inport, new InHandler {
        override def onPush(): Unit =
          if (extMessage.length > 0) {
            push(outport,extMessage)
            extMessage=""
          }
          else
            push(outport, grab(inport))
      })
      setHandler(outport, new OutHandler {
        override def onPull(): Unit = pull(inport)
      })
    }

}

参数extActor就是外部的控制Actor。在creatLogic.preStart()里我们先把StageActor传给extActor。外部系统就可以通过extActor来控制数据流行为:

class Messenger extends Actor with ActorLogging {
  var stageActor: ActorRef = _
  override def receive: Receive = {
    case r: ActorRef =>
      stageActor = r
      log.info("received stage actorRef")
    case s: String => stageActor forward s
      log.info(s"forwarding message:$s")

  }
}
object GetStageActorDemo extends App {
  implicit val sys = ActorSystem("demoSys")
  implicit val ec = sys.dispatcher
  implicit val mat = ActorMaterializer(
    ActorMaterializerSettings(sys)
      .withInputBuffer(initialSize = 16, maxSize = 16)
  )
  

  val stageActorMessenger = sys.actorOf(Props[Messenger],"forwarder")

  val source = Source(Stream.from(1)).map(_.toString).delay(1.second,DelayOverflowStrategy.backpressure)
  val graph = new StageAsActor(stageActorMessenger)
  val flow = Flow.fromGraph(graph)

  source.via(flow).to(Sink.foreach(println)).run()

   Thread.sleep(2000)
  stageActorMessenger ! "Hello"
  Thread.sleep(1000)
  stageActorMessenger ! "World!"
  Thread.sleep(2000)
  stageActorMessenger ! "Stop"
  

  scala.io.StdIn.readLine()
  sys.terminate()
  

}

Messenger就是一个存粹的中介,把控制消息通过StageActor转发给运行中的数据流。

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

GetAsyncCallBack.scala

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.stream.stage._
import scala.concurrent.duration._

//external system
object Injector {
  var callback: AsyncCallback[String] = null
   def inject(m: String) = {
     if (callback != null)
     callback.invoke(m)
   }
}
class InjectControl(injector: Injector.type) extends GraphStage[FlowShape[String,String]] {
  val inport = Inlet[String]("input")
  val outport = Outlet[String]("output")
  val shape = FlowShape.of(inport,outport)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) {
      var extMessage = ""
      override def preStart(): Unit = {
        val callback = getAsyncCallback[String] { m =>
          if (m.length > 0)
             m match {
               case "Stop" => completeStage()
               case s: String => extMessage = s
             }

        }
        injector.callback = callback
      }

      setHandler(inport, new InHandler {
        override def onPush(): Unit =
          if (extMessage.length > 0) {
            push(outport,extMessage)
            extMessage=""
          }
        else
          push(outport, grab(inport))
      })
      setHandler(outport, new OutHandler {
        override def onPull(): Unit = pull(inport)
      })
    }

}

object GetAsyncCallbackDemo extends App {
  implicit val sys = ActorSystem("demoSys")
  implicit val ec = sys.dispatcher
  implicit val mat = ActorMaterializer(
    ActorMaterializerSettings(sys)
      .withInputBuffer(initialSize = 16, maxSize = 16)
  )

  val source = Source(Stream.from(1)).map(_.toString).delay(1.second,DelayOverflowStrategy.backpressure)
  val graph = new InjectControl(Injector)
  val flow = Flow.fromGraph(graph)

  source.via(flow).to(Sink.foreach(println)).run()
  Thread.sleep(2000)
  Injector.inject("hello")
  Thread.sleep(2000)
  Injector.inject("world!")
  Thread.sleep(2000)
  Injector.inject("Stop")

  scala.io.StdIn.readLine()
  sys.terminate()


}

GetStageActorDemo.scala

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.stream.stage._
import scala.concurrent.duration._


class StageAsActor(extActor: ActorRef) extends GraphStage[FlowShape[String,String]] {
  val inport = Inlet[String]("input")
  val outport = Outlet[String]("output")
  val shape = FlowShape.of(inport,outport)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) {
      var extMessage = ""
      override def preStart(): Unit = {
        extActor ! getStageActor(behavior).ref
      }

      def behavior(m:(ActorRef,Any)): Unit = {
        val (sender, msg) = m

        msg.asInstanceOf[String] match {
          case "Stop" => completeStage()
          case s@ _ => extMessage = s
        }
      }

      setHandler(inport, new InHandler {
        override def onPush(): Unit =
          if (extMessage.length > 0) {
            push(outport,extMessage)
            extMessage=""
          }
          else
            push(outport, grab(inport))
      })
      setHandler(outport, new OutHandler {
        override def onPull(): Unit = pull(inport)
      })
    }

}

class Messenger extends Actor with ActorLogging {
  var stageActor: ActorRef = _
  override def receive: Receive = {
    case r: ActorRef =>
      stageActor = r
      log.info("received stage actorRef")
    case s: String => stageActor forward s
      log.info(s"forwarding message:$s")

  }
}
object GetStageActorDemo extends App {
  implicit val sys = ActorSystem("demoSys")
  implicit val ec = sys.dispatcher
  implicit val mat = ActorMaterializer(
    ActorMaterializerSettings(sys)
      .withInputBuffer(initialSize = 16, maxSize = 16)
  )


  val stageActorMessenger = sys.actorOf(Props[Messenger],"forwarder")

  val source = Source(Stream.from(1)).map(_.toString).delay(1.second,DelayOverflowStrategy.backpressure)
  val graph = new StageAsActor(stageActorMessenger)
  val flow = Flow.fromGraph(graph)

  source.via(flow).to(Sink.foreach(println)).run()

   Thread.sleep(2000)
  stageActorMessenger ! "Hello"
  Thread.sleep(1000)
  stageActorMessenger ! "World!"
  Thread.sleep(2000)
  stageActorMessenger ! "Stop"


  scala.io.StdIn.readLine()
  sys.terminate()


}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏风口上的猪的文章

.NET面试题系列[13] - LINQ to Object

"C# 3.0所有特性的提出都是更好地为LINQ服务的" - Learning Hard

892
来自专栏Jed的技术阶梯

Kafka 中使用 Avro 序列化框架(一):使用传统的 avro API 自定义序列化类和反序列化类

关于 avro 的 maven 工程的搭建以及 avro 的入门知识,可以参考: Apache Avro 入门

3063
来自专栏积累沉淀

初识HtmlParser

1、概念 网页解析,即程序自动分析网页内容、获取信息,从而进一步处理信息。 htmlparser包提供方便、简洁的处理html文件的方法,它将html页面中...

2055
来自专栏码匠的流水账

聊聊storm tuple的序列化

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java

1182
来自专栏码匠的流水账

聊聊rocketmq的RollingFileAppender

org/apache/rocketmq/logging/inner/LoggingBuilder.java

913
来自专栏后端之路

Dubbo自定义异常message过长解决

参考问题Dubbo异常处理 由于dubbo会将自定义异常或者第三方异常包装直接放入RuntimeException,并且使用了 StringUtils.toSt...

44110
来自专栏pangguoming

JDBC上关于数据库中多表操作一对多关系和多对多关系的实现方法

我们知道,在设计一个Java bean的时候,要把这些BEAN 的数据存放在数据库中的表结构,然而这些数据库中的表直接又有些特殊的关系,例如员工与部门直接有一对...

6837
来自专栏禁心尽力

反射+自定义注解---实现Excel数据列属性和JavaBean属性的自动映射

简单粗暴,直奔主题。 需求:通过自定义注解和反射技术,将Excel文件中的数据自动映射到pojo类中,最终返回一个List<pojo>集合?   今天我只是通...

3579
来自专栏Java成神之路

Java_脚本引擎_01_用法入门

最近有个需求,需要在js中调用java,这样能避免更新java,从而实现代码的热更新。

1064
来自专栏Hongten

spring开发_spring+hibernate

http://www.cnblogs.com/hongten/gallery/image/112469.html

933

扫码关注云+社区