标签: actor

我创建了一个名为LogActor的akka​​ actor.LogActors的receive方法处理来自其他actor的消息并将它们记录到指定的日志级别.

我可以通过两种方式区分不同的关卡.第一个:

import LogLevel._
object LogLevel extends Enumeration {
    type LogLevel = Value
    val Error, Warning, Info, Debug = Value
}
case class LogMessage(level : LogLevel, msg : String) 

第二个:( 编辑)

abstract class LogMessage(msg : String)
case class LogMessageError(msg : String) extends LogMessage(msg)
case class LogMessageWarning(msg : String) extends LogMessage(msg)
case class LogMessageInfo(msg : String) extends LogMessage(msg)
case class LogMessageDebug(msg : String) extends LogMessage(msg)

哪种方式更有效？是否需要更少的时间来匹配案例类或匹配枚举值？

(我读过这个问题,但没有任何答案提到运行时问题)

我正在学习分布式系统课程,我们必须使用Scala制作项目.我们的讲师告诉我们,Scala在使用多个内核进行计算并使用并行性来解决问题同时与actor模型集成的意义上是好的.

这是一个理论问题.我已经学习了一些关于使用Akka的actor模型的基础知识,我的问题是,在编程时,用户是否必须向编译器提供详细信息,以便各个actor在多个内核上工作,或者Scala是否会处理它并使用多个各种演员的核心？

简而言之,我的问题是:当我们在Scala中使用Akka库声明多个actor时,Scala编译器是否会自动使用多核CPU功能在内核之间分配各种actor,或者程序员是否必须提供一些输入才能执行此操作？

我有一个使用全局线程池的现有java/scala应用程序.我想开始在项目中使用actor,但希望应用程序中的所有内容都使用相同的池.

我知道我可以设置actor使用但希望共享线程池的最大线程数.这是否必要/合理,是否可以指定actor的线程池？

如果不可能/推荐,在已经使用线程的应用程序中集成actor时是否有任何经验法则？

谢谢.

有没有办法指定一个actor可以接受什么类型的消息,如果有什么东西试图发送一些其他类型的话,会给出编译错误？

在以下文档中,事件处理程序被描述为取代记录 http://akka.io/docs/akka/1.2/general/event-handler.html

有一个事件处理程序取代了Akka的日志系统:

akka.event.EventHandler

具体来说,此链接提供了使用slf4j时如何执行此操作的示例:http://akka.io/docs/akka/1.2/general/slf4j.html

我的问题是'这给了什么优势？"为什么我会这样做而不是仅使用标准模式的记录器？"

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
...
private static Logger log = LoggerFactory.getLogger(MyActor.class);
...
log.info("doing something");

基于线程或调度程序内部,通过使用上述记录器模式的事件处理程序,我是否会获得某些潜在的好处,我没有看到？如果没有,使用事件处理程序进行日志记录感觉就像偏离熟悉的模式一样没有明确的原因.

感谢您的任何意见!

我需要一个演员每分钟发送一条消息.我如何才能最好地实现这种行为？java.lang.Thread.sleep(long millis)据我所知,我害怕作为一个线程可以在Scala中的许多actor之间共享.

我在Akka文档中读到,关闭来自封闭演员的变量是危险的.

警告

在这种情况下,您需要小心避免关闭包含actor的引用,即不要在匿名Actor类中调用封闭actor上的方法.这会打破actor封装,并可能引入同步错误和竞争条件,因为其他actor的代码将同时调度到封闭的actor.

现在,我有两个演员,其中一个从第二个请求某些东西并对结果做了一些事情.在下面的这个示例中,我放在一起,演员Accumulator从演员NumberGenerator中检索数字并将它们相加,并沿途报告总和.

这可以通过至少两种不同的方式完成,因为该示例显示具有两个不同的接收功能(A对B).两者之间的区别在于A不会关闭计数器变量; 相反,它等待一个整数并将其求和,而B创建一个关闭计数器并完成总和的Future.如果我正确理解这是如何工作的,那么这是在为处理onSuccess而创建的匿名actor中发生的.

import com.esotericsoftware.minlog.Log

import akka.actor.{Actor, Props}
import akka.pattern.{ask, pipe}
import akka.util.Timeout
import akka.util.duration._

case object Start
case object Request


object ActorTest {
  var wake = 0

  val accRef = Main.actorSystem.actorOf(Props[Accumulator], name = "accumulator")
  val genRef = Main.actorSystem.actorOf(Props[NumberGenerator], name = "generator")

  Log.info("ActorTest", "Starting !")

  accRef ! Start
}

class Accumulator extends Actor {
  var …

我有一个共享的外部资源(比如一个文件存储),一个actor正在使用它.每次向文件存储器发出新请求时,都会创建一个新的actor,以通过对传入的外部系统的引用来填充请求.

我为每个参与者创建断路器的当前方法失败了目的,因为为每个"请求"创建了一个新的参与者,该请求在该外部资源上执行一系列操作.

不理想 - CB实例太多;

class MySharedResourceActor(externalResourceRef: ExtSystem) extends Actor with ActorLogging {
   val breaker = new CircuitBreaker(context.system.scheduler,
     maxFailures = 5,
     callTimeout = 10.seconds,
     resetTimeout = 1.minute).onOpen(notifyMeOnOpen())

   def receive = {
      case SomeExternalOp =>
          breaker.withSyncCircuitBreaker(dangerousCallToExternalSystem()) pipeTo sender()
   }
} 

更好的方法 - 传入CB参考;

class MySharedResourceActor(externalResourceRef: ExtSystem, val breaker: CircuitBreaker) extends Actor with ActorLogging {
   def receive = {
      case SomeExternalOp =>
          breaker.withSyncCircuitBreaker(dangerousCallToExternalSystem()) pipeTo sender()
   }
} 

传递来自父actor的断路器参考是否安全,该参与者还保持对外部系统的引用并在路由器池中的多个actor之间共享该断路器,动态创建或以其他方式？

我知道这是一个非常不精确的问题,可能被认为不适合stackoverflow.不幸的是,较小的应用程序(就演员数量而言)和"类似教程"的应用程序无法帮助我发展关于消息调度开销的直觉以及"scala对象"和"CORBA对象"之间的粒度的快速点.

虽然几乎肯定保持与客户的对话状态,例如值得一个演员,但在大多数实际使用案例中,它将涉及由许多类建模的条件/并行/替代交互.这样就可以选择将actor视为非常复杂的服务的外观,类似于刚刚退役的EJB,或者类似于smalltalk对象,只要通信可能以异步方式实现,就可以在相互之间发送消息.

除了消息传递本身的开销之外,还会有生命周期管理所涉及的开销,并且我对由于根目录中的异常或其他错误导致整个子树的链接重新启动而导致的潜在问题持谨慎态度.

为了这个问题,我们可以假设绝大多数通信发生在一台机器上,并且网络交叉是微不足道的.

我有一个长时间消耗迁移的问题,我希望在并行运行(它可以在并行运行).实际上,迁移是关于获取数据库中的所有记录,并对每个记录执行耗费时间和资源的操作.

有时单个记录迁移挂出,所以我给10分钟完成.如果迁移没有完成,我希望它正常关闭,没有任何例外(见下文)

我也使用poolboy erlang包来并行化实现,因为迁移不仅消耗时间,而且消耗资源.问题是我不知道如何在发生超时并且代码将要中断时处理错误.我的监督树是:

defmodule MyReelty.Repo.Migrations.MoveVideosFromVimeoToB2 do
  use Ecto.Migration

  alias MyReelty.Repo
  alias MyReelty.Repo.Migrations.MoveVideosFromVimeoToB2.Migrator

  # parallel nature of migration force us to disable transaction
  @disable_ddl_transaction true

  @migrator_waiting_time 10 * 60 * 1000 # timeout
  @poolboy_waiting_time @migrator_waiting_time + 10 * 1000 # give a time for graceful shutdown

  @pool_name :migrator
  @pool_size 3
  @pool_config [
    { :name, { :local, @pool_name }},
    { :worker_module, Migrator },
    { :size, @pool_size },
    { :max_overflow, 0 },
    { :strategy, :fifo }
  ]

  def up do
    children …