如何在Play!框架中通过WebSocket向Actor发送消息？

这个问题确实有点绕，Play的官方示例大多只展示了前端到Actor的单向消息流，要实现控制器主动推送消息，核心是需要集中管理所有活跃的WebSocket连接Actor，通过一个中间Actor来转发控制器的消息。下面是具体的实现步骤：

1. 创建连接管理Actor（ConnectionManager） #

首先我们需要一个单例Actor来跟踪所有在线的WebSocket连接，负责注册、注销连接，以及转发控制器的消息：

import akka.actor._

object ConnectionManager {
  // 定义消息协议
  case class RegisterConnection(actor: ActorRef)
  case class UnregisterConnection(actor: ActorRef)
  case class SendMessageToConnection(actor: ActorRef, message: String)
  case class BroadcastMessage(message: String)

  def props() = Props(new ConnectionManager())
}

class ConnectionManager extends Actor {
  import ConnectionManager._

  // 存储所有活跃的WebSocket Actor引用
  private var connections = Set.empty[ActorRef]

  def receive = {
    case RegisterConnection(actor) =>
      connections += actor
      context.watch(actor) // 监听Actor生命周期，自动处理断开连接
    case UnregisterConnection(actor) =>
      connections -= actor
    case Terminated(actor) => // 当WebSocket Actor停止时自动触发
      connections -= actor
    case SendMessageToConnection(actor, message) =>
      actor ! message // 转发消息给指定Actor
    case BroadcastMessage(message) =>
      connections.foreach(_ ! message) // 广播消息给所有连接
  }
}

2. 修改WebSocket Actor，添加注册/注销逻辑 #

修改原来的MyWebSocketActor，让它在启动时向ConnectionManager注册自己，停止时注销，同时新增接收控制器消息并转发给前端的逻辑：

import akka.actor._
import ConnectionManager._

object MyWebSocketActor {
  // 新增ConnectionManager参数，用于注册连接
  def props(out: ActorRef, connectionManager: ActorRef) = Props(new MyWebSocketActor(out, connectionManager))
}

class MyWebSocketActor(out: ActorRef, connectionManager: ActorRef) extends Actor {
  // 启动时注册当前Actor到连接管理器
  override def preStart(): Unit = {
    super.preStart()
    connectionManager ! RegisterConnection(self)
  }

  // 停止时从连接管理器注销
  override def postStop(): Unit = {
    super.postStop()
    connectionManager ! UnregisterConnection(self)
  }

  def receive = {
    // 处理前端发来的消息（保留原逻辑）
    case msg: String =>
      out ! ("I received your message: " + msg)
    // 处理控制器发来的消息，转发给前端
    case controllerMsg: String =>
      out ! controllerMsg
  }
}

3. 修改控制器，添加消息发送逻辑 #

在控制器中创建ConnectionManager的单例实例，修改WebSocket路由的创建逻辑，同时新增处理按钮点击的Action来发送消息：

import play.api.mvc._
import play.api.libs.streams.ActorFlow
import javax.inject.Inject
import akka.actor.{ActorSystem, ActorRef, Props}
import akka.stream.Materializer
import ConnectionManager._

class Application @Inject()(cc: ControllerComponents)(implicit system: ActorSystem, mat: Materializer) extends AbstractController(cc) {
  // 初始化单例连接管理器
  private val connectionManager: ActorRef = system.actorOf(ConnectionManager.props(), "connection-manager")

  // 修改WebSocket创建逻辑，传入连接管理器
  def socket = WebSocket.accept[String, String] { request =>
    ActorFlow.actorRef { out =>
      MyWebSocketActor.props(out, connectionManager)
    }
  }

  // 处理按钮点击的Action，这里以广播消息为例
  def sendMessageToWebSocket = Action { implicit request =>
    // 方案1：广播消息给所有在线WebSocket连接
    connectionManager ! BroadcastMessage("Hello from controller!")

    // 方案2：发送给指定连接（需要你自己维护用户与Actor的映射，比如从Session取用户ID）
    // val userId = request.session.get("userId").getOrElse("")
    // connectionManager ! SendMessageToUser(userId, "Hello to specific user!")

    Ok("Message sent to WebSocket connections")
  }
}

关键说明 #

• 连接管理：ConnectionManager作为中间层，解耦了控制器和WebSocket Actor，避免控制器直接持有大量Actor引用。
• 特定用户推送：如果需要给指定用户发送消息，可以扩展ConnectionManager的消息协议，比如新增RegisterConnection(userId: String, actor: ActorRef)，用Map[String, ActorRef]存储用户ID与Actor的映射，控制器就可以根据用户ID精准推送。
• 生命周期管理：通过context.watch(actor)监听Actor的停止事件，确保连接列表始终是最新的，避免发送消息到已断开的连接。

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