Akka(43): Http:SSE-Server Sent Event - 服务端主推消息

2024-04-09 04:48

本文主要是介绍Akka(43): Http:SSE-Server Sent Event - 服务端主推消息,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

   因为我了解Akka-http的主要目的不是为了有关Web-Server的编程,而是想实现一套系统集成的api,所以也需要考虑由服务端主动向客户端发送指令的应用场景。比如一个零售店管理平台的服务端在完成了某些数据更新后需要通知各零售门市客户端下载最新数据。虽然Akka-http也提供对websocket协议的支持,但websocket的网络连接是双向恒久的,适合频繁的问答交互式服务端与客户端的交流,消息结构也比较零碎。而我们面临的可能是批次型的大量数据库数据交换,只需要简单的服务端单向消息就行了,所以websocket不太合适,而Akka-http的SSE应该比较适合我们的要求。SSE模式的基本原理是服务端统一集中发布消息,各客户端持久订阅服务端发布的消息并从消息的内容中筛选出属于自己应该执行的指令,然后进行相应的处理。客户端接收SSE是在一个独立的线程里不断进行的,不会影响客户端当前的运算流程。当收到有用的消息后就会调用一个业务功能函数作为后台异步运算任务。

服务端的SSE发布是以Source[ServerSentEvent,NotUsed]来实现的。ServerSentEvent类型定义如下:

/*** Representation of a server-sent event. According to the specification, an empty data field designates an event* which is to be ignored which is useful for heartbeats.** @param data data, may span multiple lines* @param eventType optional type, must not contain \n or \r* @param id optional id, must not contain \n or \r* @param retry optional reconnection delay in milliseconds*/
final case class ServerSentEvent(data:      String,eventType: Option[String] = None,id:        Option[String] = None,retry:     Option[Int]    = None) {...}

这个类型的参数代表事件消息的数据结构。用户可以根据实际需要充分利用这个数据结构来传递消息。服务端是通过complete以SeverSentEvent类为元素 的Source来进行SSE的,如下:

    import akka.http.scaladsl.marshalling.sse.EventStreamMarshalling._complete {Source.tick(2.seconds, 2.seconds, NotUsed).map( _ => processToServerSentEvent).keepAlive(1.second, () => ServerSentEvent.heartbeat)}

以上代码代表服务端定时运算processToServerSentEvent返回ServerSentEvent类型结果后发布给所有订阅的客户端。我们用一个函数processToServerSentEvent模拟重复运算的业务功能:

  private def processToServerSentEvent: ServerSentEvent = {Thread.sleep(3000)   //processing delayServerSentEvent(SyncFiles.fileToSync)}

这个函数模拟发布事件数据是某种业务运算结果,在这里代表客户端需要下载文件名称。我们用客户端request来模拟设定这个文件名称:

  object SyncFiles {var fileToSync: String = ""}private def route = {import Directives._import akka.http.scaladsl.marshalling.sse.EventStreamMarshalling._def syncRequests =pathPrefix("sync") {pathSingleSlash {post {parameter("file") { filename =>complete {SyncFiles.fileToSync = filenames"set download file to : $filename"}}}}}

客户端订阅SSE的方式如下:

    import akka.http.scaladsl.unmarshalling.sse.EventStreamUnmarshalling._import system.dispatcherHttp().singleRequest(Get("http://localhost:8011/events")).flatMap(Unmarshal(_).to[Source[ServerSentEvent, NotUsed]]).foreach(_.runForeach(se => downloadFiles(se.data)))

每当客户端收到SSE后即运行downloadFiles(filename)函数。downloadFiles函数定义:

  def downloadFiles(file: String) = {Thread.sleep(3000)   //process delayif (file != "")println(s"Try to download $file")}

下面是客户端程序的测试运算步骤:

    scala.io.StdIn.readLine()println("do some thing ...")Http().singleRequest(HttpRequest(method=HttpMethods.POST,uri = "http://localhost:8011/sync/?file=Orders")).onSuccess {case msg => println(msg)}scala.io.StdIn.readLine()println("do some other things ...")Http().singleRequest(HttpRequest(method=HttpMethods.POST,uri = "http://localhost:8011/sync/?file=Items")).onSuccess {case msg => println(msg)}

运算结果:

do some thing ...
HttpResponse(200 OK,List(Server: akka-http/10.0.10, Date: Fri, 15 Dec 2017 05:50:52 GMT),HttpEntity.Strict(text/plain; charset=UTF-8,set download file to : Orders),HttpProtocol(HTTP/1.1))
Try to download Orders
Try to download Ordersdo some other things ...
HttpResponse(200 OK,List(Server: akka-http/10.0.10, Date: Fri, 15 Dec 2017 05:51:02 GMT),HttpEntity.Strict(text/plain; charset=UTF-8,set download file to : Items),HttpProtocol(HTTP/1.1))
Try to download Orders
Try to download Orders
Try to download Items
Try to download ItemsTry to download ItemsProcess finished with exit code 0

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

服务端:

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import scala.concurrent.duration.DurationInt
import akka.http.scaladsl.model.sse.ServerSentEventobject SSEServer {def main(args: Array[String]): Unit = {implicit val system = ActorSystem()implicit val mat    = ActorMaterializer()Http().bindAndHandle(route, "localhost", 8011)scala.io.StdIn.readLine()system.terminate()}object SyncFiles {var fileToSync: String = ""}private def route = {import Directives._import akka.http.scaladsl.marshalling.sse.EventStreamMarshalling._def syncRequests =pathPrefix("sync") {pathSingleSlash {post {parameter("file") { filename =>complete {SyncFiles.fileToSync = filenames"set download file to : $filename"}}}}}def events =path("events") {get {complete {Source.tick(2.seconds, 2.seconds, NotUsed).map( _ => processToServerSentEvent).keepAlive(1.second, () => ServerSentEvent.heartbeat)}}}syncRequests ~ events}private def processToServerSentEvent: ServerSentEvent = {Thread.sleep(3000)   //processing delayServerSentEvent(SyncFiles.fileToSync)}
}

客户端:

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuilding.Get
import akka.http.scaladsl.model.HttpMethods
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import akka.http.scaladsl.model.sse.ServerSentEvent
import akka.http.scaladsl.model._object SSEClient {def downloadFiles(file: String) = {Thread.sleep(3000)   //process delayif (file != "")println(s"Try to download $file")}def main(args: Array[String]): Unit = {implicit val system = ActorSystem()implicit val mat    = ActorMaterializer()import akka.http.scaladsl.unmarshalling.sse.EventStreamUnmarshalling._import system.dispatcherHttp().singleRequest(Get("http://localhost:8011/events")).flatMap(Unmarshal(_).to[Source[ServerSentEvent, NotUsed]]).foreach(_.runForeach(se => downloadFiles(se.data)))scala.io.StdIn.readLine()println("do some thing ...")Http().singleRequest(HttpRequest(method=HttpMethods.POST,uri = "http://localhost:8011/sync/?file=Orders")).onSuccess {case msg => println(msg)}scala.io.StdIn.readLine()println("do some other things ...")Http().singleRequest(HttpRequest(method=HttpMethods.POST,uri = "http://localhost:8011/sync/?file=Items")).onSuccess {case msg => println(msg)}scala.io.StdIn.readLine()system.terminate()}
}






这篇关于Akka(43): Http:SSE-Server Sent Event - 服务端主推消息的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/887211

相关文章

Spring Boot 整合 SSE的高级实践(Server-Sent Events)

《SpringBoot整合SSE的高级实践(Server-SentEvents)》SSE(Server-SentEvents)是一种基于HTTP协议的单向通信机制,允许服务器向浏览器持续发送实... 目录1、简述2、Spring Boot 中的SSE实现2.1 添加依赖2.2 实现后端接口2.3 配置超时时

Redis消息队列实现异步秒杀功能

《Redis消息队列实现异步秒杀功能》在高并发场景下,为了提高秒杀业务的性能,可将部分工作交给Redis处理,并通过异步方式执行,Redis提供了多种数据结构来实现消息队列,总结三种,本文详细介绍Re... 目录1 Redis消息队列1.1 List 结构1.2 Pub/Sub 模式1.3 Stream 结

springboot整合阿里云百炼DeepSeek实现sse流式打印的操作方法

《springboot整合阿里云百炼DeepSeek实现sse流式打印的操作方法》:本文主要介绍springboot整合阿里云百炼DeepSeek实现sse流式打印,本文给大家介绍的非常详细,对大... 目录1.开通阿里云百炼,获取到key2.新建SpringBoot项目3.工具类4.启动类5.测试类6.测

SQL server配置管理器找不到如何打开它

《SQLserver配置管理器找不到如何打开它》最近遇到了SQLserver配置管理器打不开的问题,尝试在开始菜单栏搜SQLServerManager无果,于是将自己找到的方法总结分享给大家,对SQ... 目录方法一:桌面图标进入方法二:运行窗口进入方法三:查找文件路径方法四:检查 SQL Server 安

python连接本地SQL server详细图文教程

《python连接本地SQLserver详细图文教程》在数据分析领域,经常需要从数据库中获取数据进行分析和处理,下面:本文主要介绍python连接本地SQLserver的相关资料,文中通过代码... 目录一.设置本地账号1.新建用户2.开启双重验证3,开启TCP/IP本地服务二js.python连接实例1.

在Android平台上实现消息推送功能

《在Android平台上实现消息推送功能》随着移动互联网应用的飞速发展,消息推送已成为移动应用中不可或缺的功能,在Android平台上,实现消息推送涉及到服务端的消息发送、客户端的消息接收、通知渠道(... 目录一、项目概述二、相关知识介绍2.1 消息推送的基本原理2.2 Firebase Cloud Me

Nginx中配置HTTP/2协议的详细指南

《Nginx中配置HTTP/2协议的详细指南》HTTP/2是HTTP协议的下一代版本,旨在提高性能、减少延迟并优化现代网络环境中的通信效率,本文将为大家介绍Nginx配置HTTP/2协议想详细步骤,需... 目录一、HTTP/2 协议概述1.HTTP/22. HTTP/2 的核心特性3. HTTP/2 的优

使用Python自建轻量级的HTTP调试工具

《使用Python自建轻量级的HTTP调试工具》这篇文章主要为大家详细介绍了如何使用Python自建一个轻量级的HTTP调试工具,文中的示例代码讲解详细,感兴趣的小伙伴可以参考一下... 目录一、为什么需要自建工具二、核心功能设计三、技术选型四、分步实现五、进阶优化技巧六、使用示例七、性能对比八、扩展方向建

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》:本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优