Akka（36）： Http：Client-side-Api，Client-Connections

2024-04-09 04:48
文章标签 36 http api client akka side connections

本文主要是介绍Akka（36）： Http：Client-side-Api，Client-Connections，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

   Akka-http的客户端Api应该是以HttpRequest操作为主轴的网上消息交换模式编程工具。我们知道：Akka-http是搭建在Akka-stream之上的。所以，Akka-http在客户端构建与服务器的连接通道也可以用Akka-stream的Flow来表示。这个Flow可以通过调用Http.outgoingConnection来获取：

  /*** Creates a [[akka.stream.scaladsl.Flow]] representing a prospective HTTP client connection to the given endpoint.* Every materialization of the produced flow will attempt to establish a new outgoing connection.** To configure additional settings for requests made using this method,* use the `akka.http.client` config section or pass in a [[akka.http.scaladsl.settings.ClientConnectionSettings]] explicitly.*/def outgoingConnection(host: String, port: Int = 80,localAddress: Option[InetSocketAddress] = None,settings:     ClientConnectionSettings  = ClientConnectionSettings(system),log:          LoggingAdapter            = system.log): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] =_outgoingConnection(host, port, settings.withLocalAddressOverride(localAddress), ConnectionContext.noEncryption(), ClientTransport.TCP, log)

我们看到：这个函数实现了对Server端地址host+port的设定，返回的结果类型是Flow[HttpRequest,HttpResponse,Future[OutgoingConnection]]。这个Flow代表将输入的HttpRequest转换成输出的HttpResponse。这个转换过程包括了与Server之间的Request,Response消息交换。下面我们试着用这个Flow来向Server端发送request，并获取response： 

  val connFlow: Flow[HttpRequest,HttpResponse,Future[Http.OutgoingConnection]] =Http().outgoingConnection("akka.io")def sendHttpRequest(req: HttpRequest) = {Source.single(req).via(connFlow).runWith(Sink.head)}sendHttpRequest(HttpRequest(uri="/")).andThen{case Success(resp) => println(s"got response: ${resp.status.intValue()}")case Failure(err) => println(s"request failed: ${err.getMessage}")}.andThen {case _ => sys.terminate()}

以上用法只能处理零星短小的requests，这是因为虽然connFlow是一次性实例化，但每次调用runWith都会构建新的connection，而实例化和构建新connection会拖慢系统运行速度，不适用于像streaming这样大量消息的相互传递。

上面的这种模式就是所谓Connection-Level-Client-Side-Api。这种模式可以让用户有更大程度的自由度控制connection的构建、使用及在connection上发送request的方式。一般来讲，当返回response的entity被完全消耗后系统会自动close connection，这套api还提供了一些手动方法可以在有需要的情况下手动进行connection close，如下：

 //close connection by cancelling response entityresp.entity.dataBytes.runWith(Sink.cancelled)//close connection by receiving response with close headerHttp().bindAndHandleSync({ req ⇒ HttpResponse(headers = headers.Connection("close") :: Nil) },"akka.io",80)(mat)

Akka-http客户端api还有一种实用的Host-Level-Client-Side-Api模式。这套api能自动针对每个端点维护一个连接池（connection-pool），用户只需对连接池进行配置。系统按照连接池配置自动维护池内线程的生、死、动、停。akka-http.host-connection-pool配置中max-connections,max-open-requests,pipelining-limit等控制着connection、在途request的数量，需要特别注意。针对某个端点的连接池是通过Http().cachedHostConnectionPool(endPoint)获取的。同样，获取的也是一个client-flow实例。因为系统自动维护着线程池，所以client-flow实例可以任意引用，无论调用次数与调用时间间隔。cachedHostConnectionPool()函数定义如下：

  /*** Same as [[#cachedHostConnectionPool]] but for encrypted (HTTPS) connections.** If an explicit [[ConnectionContext]] is given then it rather than the configured default [[ConnectionContext]] will be used* for encryption on the connections.** To configure additional settings for the pool (and requests made using it),* use the `akka.http.host-connection-pool` config section or pass in a [[ConnectionPoolSettings]] explicitly.*/def cachedHostConnectionPoolHttps[T](host: String, port: Int = 443,connectionContext: HttpsConnectionContext = defaultClientHttpsContext,settings:          ConnectionPoolSettings = defaultConnectionPoolSettings,log:               LoggingAdapter         = system.log)(implicit fm: Materializer): Flow[(HttpRequest, T), (Try[HttpResponse], T), HostConnectionPool] = {val cps = ConnectionPoolSetup(settings, connectionContext, log)val setup = HostConnectionPoolSetup(host, port, cps)cachedHostConnectionPool(setup)}

函数返回结果类型：Flow[(HttpRequest,T),(Try[HttpResponse],T),HostConnectionPool]。因为线程池内的线程是异步构建request和接收response的，而返回response的顺序未必按照发送request的顺序，所以需要一个tuple2的T类型标示request与返回的response进行匹配。线程池会根据idle-timeout自动终止，也可以手动通过HostConnectionPool.shutDown()实现： 

  /*** Represents a connection pool to a specific target host and pool configuration.*/final case class HostConnectionPool private[http] (setup: HostConnectionPoolSetup)(private[http] val gateway: PoolGateway) { // enable test access/*** Asynchronously triggers the shutdown of the host connection pool.** The produced [[scala.concurrent.Future]] is fulfilled when the shutdown has been completed.*/def shutdown()(implicit ec: ExecutionContextExecutor): Future[Done] = gateway.shutdown()private[http] def toJava = new akka.http.javadsl.HostConnectionPool {override def setup = HostConnectionPool.this.setupoverride def shutdown(executor: ExecutionContextExecutor): CompletionStage[Done] = HostConnectionPool.this.shutdown()(executor).toJava}}

也可以通过Http().shutdownAllConnectionPools()一次性终止ActorSystem内所有线程池: 

  /*** Triggers an orderly shutdown of all host connections pools currently maintained by the [[akka.actor.ActorSystem]].* The returned future is completed when all pools that were live at the time of this method call* have completed their shutdown process.** If existing pool client flows are re-used or new ones materialized concurrently with or after this* method call the respective connection pools will be restarted and not contribute to the returned future.*/def shutdownAllConnectionPools(): Future[Unit] = {val shutdownCompletedPromise = Promise[Done]()poolMasterActorRef ! ShutdownAll(shutdownCompletedPromise)shutdownCompletedPromise.future.map(_ ⇒ ())(system.dispatcher)}

我们用cachedHostConnectionPool获取一个client-flow实例：

Flow[(HttpRequest,T),(Try[HttpResponse],T),HostConnectionPool]后就可以进行输入HttpRequest到HttpResponse的转换处理。如下面的例子：

  val pooledFlow: Flow[(HttpRequest,Int),(Try[HttpResponse],Int),Http.HostConnectionPool] =Http().cachedHostConnectionPool[Int](host="akka.io",port=80)def sendPoolRequest(req: HttpRequest, marker: Int) = {Source.single(req -> marker).via(pooledFlow).runWith(Sink.head)}sendPoolRequest(HttpRequest(uri="/"), 1).andThen{case Success((tryResp, mk)) =>tryResp match {case Success(resp) => println(s"got response: ${resp.status.intValue()}")case Failure(err) => println(s"request failed: ${err.getMessage}")}case Failure(err) => println(s"request failed: ${err.getMessage}")}.andThen {case _ => sys.terminate()}

在以上这个例子里实际同样会遇到Connection-Level-Api所遇的的问题，这是因为获取的线程池内的线程还是有限的，只能缓解因为request速率超出response速率所造成的request积压。目前最有效的方法还是通过使用一个queue来暂存request后再逐个处理：

    val QueueSize = 10// This idea came initially from this blog post:// http://kazuhiro.github.io/scala/akka/akka-http/akka-streams/2016/01/31/connection-pooling-with-akka-http-and-source-queue.htmlval poolClientFlow = Http().cachedHostConnectionPool[Promise[HttpResponse]]("akka.io")val queue =Source.queue[(HttpRequest, Promise[HttpResponse])](QueueSize, OverflowStrategy.dropNew).via(poolClientFlow).toMat(Sink.foreach({case ((Success(resp), p)) => p.success(resp)case ((Failure(e), p))    => p.failure(e)}))(Keep.left).run()def queueRequest(request: HttpRequest): Future[HttpResponse] = {val responsePromise = Promise[HttpResponse]()queue.offer(request -> responsePromise).flatMap {case QueueOfferResult.Enqueued    => responsePromise.futurecase QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed. Try again later."))case QueueOfferResult.Failure(ex) => Future.failed(ex)case QueueOfferResult.QueueClosed => Future.failed(new RuntimeException("Queue was closed (pool shut down) while running the request. Try again later."))}}val responseFuture: Future[HttpResponse] = queueRequest(HttpRequest(uri = "/"))responseFuture.andThen {case Success(resp) => println(s"got response: ${resp.status.intValue()}")case Failure(err) => println(s"request failed: ${err.getMessage}")}.andThen {case _ => sys.terminate()}

下面是本次Akka-http-client-side-connection讨论的示范源代码： 

import akka.actor._
import akka.http.javadsl.{HostConnectionPool, OutgoingConnection}
import akka.stream._
import akka.stream.scaladsl._
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._import scala.concurrent._
import scala.util._object ClientApiDemo extends App {implicit val sys = ActorSystem("ClientSys")implicit val mat = ActorMaterializer()implicit val ec = sys.dispatcher
/*val connFlow: Flow[HttpRequest,HttpResponse,Future[Http.OutgoingConnection]] =Http().outgoingConnection("www.sina.com")def sendHttpRequest(req: HttpRequest) = {Source.single(req).via(connFlow).runWith(Sink.head)}sendHttpRequest(HttpRequest(uri="/")).andThen{case Success(resp) =>//close connection by cancelling response entityresp.entity.dataBytes.runWith(Sink.cancelled)println(s"got response: ${resp.status.intValue()}")case Failure(err) => println(s"request failed: ${err.getMessage}")}//   .andThen {case _ => sys.terminate()}//close connection by receiving response with close headerHttp().bindAndHandleSync({ req ⇒ HttpResponse(headers = headers.Connection("close") :: Nil) },"akka.io",80)(mat)val pooledFlow: Flow[(HttpRequest,Int),(Try[HttpResponse],Int),Http.HostConnectionPool] =Http().cachedHostConnectionPool[Int](host="akka.io",port=80)def sendPoolRequest(req: HttpRequest, marker: Int) = {Source.single(req -> marker).via(pooledFlow).runWith(Sink.head)}sendPoolRequest(HttpRequest(uri="/"), 1).andThen{case Success((tryResp, mk)) =>tryResp match {case Success(resp) => println(s"got response: ${resp.status.intValue()}")case Failure(err) => println(s"request failed: ${err.getMessage}")}case Failure(err) => println(s"request failed: ${err.getMessage}")}.andThen {case _ => sys.terminate()}
*/val QueueSize = 10// This idea came initially from this blog post:// http://kazuhiro.github.io/scala/akka/akka-http/akka-streams/2016/01/31/connection-pooling-with-akka-http-and-source-queue.htmlval poolClientFlow = Http().cachedHostConnectionPool[Promise[HttpResponse]]("akka.io")val queue =Source.queue[(HttpRequest, Promise[HttpResponse])](QueueSize, OverflowStrategy.dropNew).via(poolClientFlow).toMat(Sink.foreach({case ((Success(resp), p)) => p.success(resp)case ((Failure(e), p))    => p.failure(e)}))(Keep.left).run()def queueRequest(request: HttpRequest): Future[HttpResponse] = {val responsePromise = Promise[HttpResponse]()queue.offer(request -> responsePromise).flatMap {case QueueOfferResult.Enqueued    => responsePromise.futurecase QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed. Try again later."))case QueueOfferResult.Failure(ex) => Future.failed(ex)case QueueOfferResult.QueueClosed => Future.failed(new RuntimeException("Queue was closed (pool shut down) while running the request. Try again later."))}}val responseFuture: Future[HttpResponse] = queueRequest(HttpRequest(uri = "/"))responseFuture.andThen {case Success(resp) => println(s"got response: ${resp.status.intValue()}")case Failure(err) => println(s"request failed: ${err.getMessage}")}.andThen {case _ => sys.terminate()}}






这篇关于Akka（36）： Http：Client-side-Api，Client-Connections的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/887204。

相关文章

Java调用DeepSeek API的最佳实践及详细代码示例

Java调用DeepSeek API的最佳实践及详细代码示例

《Java调用DeepSeekAPI的最佳实践及详细代码示例》：本文主要介绍如何使用Java调用DeepSeekAPI,包括获取API密钥、添加HTTP客户端依赖、创建HTTP请求、处理响应、... 目录1. 获取API密钥2. 添加HTTP客户端依赖3. 创建HTTP请求4. 处理响应5. 错误处理6.
阅读更多...
Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

《DeepseekR1模型本地化部署+API接口调用详细教程(释放AI生产力)》本文介绍了本地部署DeepSeekR1模型和通过API调用将其集成到VSCode中的过程,作者详细步骤展示了如何下载和... 目录前言一、deepseek R1模型与chatGPT o1系列模型对比二、本地部署步骤1.安装oll
阅读更多...
浅析如何使用Swagger生成带权限控制的API文档

浅析如何使用Swagger生成带权限控制的API文档

《浅析如何使用Swagger生成带权限控制的API文档》当涉及到权限控制时,如何生成既安全又详细的API文档就成了一个关键问题,所以这篇文章小编就来和大家好好聊聊如何用Swagger来生成带有... 目录准备工作配置 Swagger权限控制给 API 加上权限注解查看文档注意事项在咱们的开发工作里，API
阅读更多...
一分钟带你上手Python调用DeepSeek的API

一分钟带你上手Python调用DeepSeek的API

《一分钟带你上手Python调用DeepSeek的API》最近DeepSeek非常火,作为一枚对前言技术非常关注的程序员来说,自然都想对接DeepSeek的API来体验一把,下面小编就来为大家介绍一下... 目录前言免费体验API-Key申请首次调用API基本概念最小单元推理模型智能体自定义界面总结前言最
阅读更多...
JAVA调用Deepseek的api完成基本对话简单代码示例

JAVA调用Deepseek的api完成基本对话简单代码示例

《JAVA调用Deepseek的api完成基本对话简单代码示例》：本文主要介绍JAVA调用Deepseek的api完成基本对话的相关资料,文中详细讲解了如何获取DeepSeekAPI密钥、添加H... 获取API密钥首先，从DeepSeek平台获取API密钥，用于身份验证。添加HTTP客户端依赖使用Jav
阅读更多...
C#使用DeepSeek API实现自然语言处理,文本分类和情感分析

C#使用DeepSeek API实现自然语言处理,文本分类和情感分析

《C#使用DeepSeekAPI实现自然语言处理,文本分类和情感分析》在C#中使用DeepSeekAPI可以实现多种功能,例如自然语言处理、文本分类、情感分析等,本文主要为大家介绍了具体实现步骤,... 目录准备工作文本生成文本分类问答系统代码生成翻译功能文本摘要文本校对图像描述生成总结在C#中使用Deep
阅读更多...
5分钟获取deepseek api并搭建简易问答应用

5分钟获取deepseek api并搭建简易问答应用

《5分钟获取deepseekapi并搭建简易问答应用》本文主要介绍了5分钟获取deepseekapi并搭建简易问答应用,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需... 目录1、获取api2、获取base_url和chat_model3、配置模型参数方法一：终端中临时将加
阅读更多...
使用DeepSeek API 结合VSCode提升开发效率

使用DeepSeek API 结合VSCode提升开发效率

《使用DeepSeekAPI结合VSCode提升开发效率》：本文主要介绍DeepSeekAPI与VisualStudioCode(VSCode)结合使用,以提升软件开发效率,具有一定的参考价值... 目录引言准备工作安装必要的 VSCode 扩展配置 DeepSeek API1. 创建 API 请求文件2.
阅读更多...
详解Java如何向http/https接口发出请求

详解Java如何向http/https接口发出请求

《详解Java如何向http/https接口发出请求》这篇文章主要为大家详细介绍了Java如何实现向http/https接口发出请求,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 用Java发送web请求所用到的包都在java.net下，在具体使用时可以用如下代码，你可以把它封装成一
阅读更多...
使用SpringBoot创建一个RESTful API的详细步骤

使用SpringBoot创建一个RESTful API的详细步骤

《使用SpringBoot创建一个RESTfulAPI的详细步骤》使用Java的SpringBoot创建RESTfulAPI可以满足多种开发场景,它提供了快速开发、易于配置、可扩展、可维护的优点,尤... 目录一、创建 Spring Boot 项目二、创建控制器类（Controller Class）三、运行
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2