基于 Kotlin + OkHttp 实现易用且功能强大的网络框架（一）

fengzhizi715

发布于 2021-11-24 13:02:54

2.5K0

发布于 2021-11-24 13:02:54

文章被收录于专栏：Java与Android技术栈

okhttp-extension 是针对 okhttp 3 增强的网络框架。使用 Kotlin 特性编写，提供便捷的 DSL 方式创建网络请求，支持协程、响应式编程等等。

其 core 模块只依赖 OkHttp，不会引入第三方库。

okhttp-extension 可以整合 Retrofit、Feign 框架，还提供了很多常用的拦截器。另外，okhttp-extension 也给开发者提供一种新的选择。

github地址：https://github.com/fengzhizi715/okhttp-extension

Features:

支持 DSL 创建 HTTP GET/POST/PUT/HEAD/DELETE/PATCH requests.
支持 Kotlin 协程
支持响应式(RxJava、Spring Reactor)
支持函数式
支持熔断器(Resilience4j)
支持异步请求的取消
支持 Request、Response 的拦截器
提供常用的拦截器
支持自定义线程池
支持整合 Retrofit、Feign 框架
支持 Websocket 的实现、自动重连等
core 模块只依赖 OkHttp，不依赖其他第三方库

okhttp-extension.png

一. General

1.1 Basic

无需任何配置(零配置)即可直接使用，仅限于 Get 请求。

    "https://baidu.com".httpGet().use {
        println(it)
    }

或者需要依赖协程，也仅限于 Get 请求。

   "https://baidu.com".asyncGet()
       .await()
       .use {
           println(it)
       }

1.2 Config

配置 OkHttp 相关的参数以及拦截器，例如：

const val DEFAULT_CONN_TIMEOUT = 30

val loggingInterceptor by lazy {
    LogManager.logProxy(object : LogProxy {  // 必须要实现 LogProxy ，否则无法打印网络请求的 request 、response
        override fun e(tag: String, msg: String) {
        }

        override fun w(tag: String, msg: String) {
        }

        override fun i(tag: String, msg: String) {
            println("$tag:$msg")
        }

        override fun d(tag: String, msg: String) {
            println("$tag:$msg")
        }
    })

    LoggingInterceptor.Builder()
        .loggable(true) // TODO: 发布到生产环境需要改成false
        .request()
        .requestTag("Request")
        .response()
        .responseTag("Response")
//        .hideVerticalLine()// 隐藏竖线边框
        .build()
}

val httpClient: HttpClient by lazy {
    HttpClientBuilder()
        .baseUrl("http://localhost:8080")
        .allTimeouts(DEFAULT_CONN_TIMEOUT.toLong(), TimeUnit.SECONDS)
        .addInterceptor(loggingInterceptor)
        .addInterceptor(CurlLoggingInterceptor())
        .serializer(GsonSerializer())
        .jsonConverter(GlobalRequestJSONConverter::class)
        .build()
}

配置完之后，就可以直接使用 httpClient

    httpClient.get{

        url {
            url = "/response-headers-queries"

            "param1" to "value1"
            "param2" to "value2"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }
    }.use {
        println(it)
    }

这里的 url 需要和 baseUrl 组成完整的 url。比如：http://localhost:8080/response-headers-queries 当然，也可以使用 customUrl 替代 baseUrl + url 作为完整的 url

1.3 AOP

针对所有 request、response 做一些类似 AOP 的行为。

需要在构造 httpClient 时，调用 addRequestProcessor()、addResponseProcessor() 方法，例如：

val httpClientWithAOP by lazy {
    HttpClientBuilder()
        .baseUrl("http://localhost:8080")
        .allTimeouts(DEFAULT_CONN_TIMEOUT.toLong(), TimeUnit.SECONDS)
        .addInterceptor(loggingInterceptor)
        .serializer(GsonSerializer())
        .jsonConverter(GlobalRequestJSONConverter::class)
        .addRequestProcessor { _, builder ->
            println("request start")
            builder
        }
        .addResponseProcessor {
            println("response start")
        }
        .build()
}

这样在进行 request、response 时，会分别打印"request start"和"response start"。

因为在创建 request 之前，会处理所有的 RequestProcessor；在响应 response 之前，也会用内部的 ResponseProcessingInterceptor 拦截器来处理 ResponseProcessor。

RequestProcessor、ResponseProcessor 分别可以认为是 request、response 的拦截器。

// a request interceptor
typealias RequestProcessor = (HttpClient, Request.Builder) -> Request.Builder

// a response interceptor
typealias ResponseProcessor = (Response) -> Unit

我们可以多次调用 addRequestProcessor() 、addResponseProcessor() 方法。

二. DSL

DSL 是okhttp-extension框架的特色。包含使用 DSL 创建各种 HTTP Request 和使用 DSL 结合声明式编程。

2.1 HTTP Request

使用 DSL 支持创建GET/POST/PUT/HEAD/DELETE/PATCH

2.1.1 get

最基本的 get 用法

    httpClient.get{

        url {
            url = "/response-headers-queries"

            "param1" to "value1"
            "param2" to "value2"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }
    }.use {
        println(it)
    }

这里的 url 需要和 baseUrl 组成完整的 url。比如：http://localhost:8080/response-headers-queries 当然，也可以使用 customUrl 替代 baseUrl + url 作为完整的 url

2.1.2 post

基本的 post 请求如下：

    httpClient.post{

        url {
            url = "/response-body"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }

        body {
            form {
                "form1" to "value1"
                "form2" to "value2"
            }
        }
    }.use {
        println(it)
    }

支持 request body 为 json 字符串

    httpClient.post{

        url {
            url = "/response-body"
        }

        body("application/json") {
            json {
                "key1" to "value1"
                "key2" to "value2"
                "key3" to "value3"
            }
        }
    }.use {
        println(it)
    }

支持单个/多个文件的上传

    val file = File("/Users/tony/Downloads/xxx.png")

    httpClient.post{

        url {
            url = "/upload"
        }

        multipartBody {
            +part("file", file.name) {
                file(file)
            }
        }
    }.use {
        println(it)
    }

更多 post 相关的方法，欢迎使用者自行探索。

2.1.3 put

    httpClient.put{

        url {
            url = "/response-body"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }

        body("application/json") {
            string("content")
        }
    }.use {
        println(it)
    }

2.1.4 delete

    httpClient.delete{

        url {
            url = "/users/tony"
        }
    }.use {
        println(it)
    }

2.1.5 head

    httpClient.head{

        url {
            url = "/response-headers"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
            "key3" to "value3"
        }
    }.use {
        println(it)
    }

2.1.6 patch

    httpClient.patch{

        url {
            url = "/response-body"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }

        body("application/json") {
            string("content")
        }
    }.use {
        println(it)
    }

2.2 Declarative

像使用 Retrofit、Feign 一样，在配置完 httpClient 之后，需要定义一个 ApiService 它用于声明所调用的全部接口。ApiService 所包含的方法也是基于 DSL 的。例如：

class ApiService(client: HttpClient) : AbstractHttpService(client) {

    fun testGet(name: String) = get<Response> {
        url = "/sayHi/$name"
    }

    fun testGetWithPath(path: Map<String, String>) = get<Response> {
        url = "/sayHi/{name}"
        pathParams = Params.from(path)
    }

    fun testGetWithHeader(headers: Map<String, String>) = get<Response> {
        url = "/response-headers"
        headersParams = Params.from(headers)
    }

    fun testGetWithHeaderAndQuery(headers: Map<String, String>, queries: Map<String,String>) = get<Response> {
        url = "/response-headers-queries"
        headersParams = Params.from(headers)
        queriesParams = Params.from(queries)
    }

    fun testPost(body: Params) = post<Response> {
        url = "/response-body"
        bodyParams = body
    }

    fun testPostWithModel(model: RequestModel) = post<Response>{
        url = "/response-body"
        bodyModel = model
    }

    fun testPostWithJsonModel(model: RequestModel) = jsonPost<Response>{
        url = "/response-body-with-model"
        jsonModel = model
    }

    fun testPostWithResponseMapper(model: RequestModel) = jsonPost<ResponseData>{
        url = "/response-body-with-model"
        jsonModel = model
        responseMapper = ResponseDataMapper::class
    }
}

定义好 ApiService 就可以直接使用了，例如：

val apiService by lazy {
    ApiService(httpClient)
}

val requestModel = RequestModel()
apiService.testPostWithModel(requestModel).sync()

当然也支持异步，会返回CompletableFuture对象，例如：

val apiService by lazy {
    ApiService(httpClient)
}

val requestModel = RequestModel()
apiService.testPostWithModel(requestModel).async()

借助于 Kotlin 扩展函数的特性，也支持返回 RxJava 的 Observable 对象等、Reactor 的 Flux/Mono 对象、Kotlin Coroutines 的 Flow 对象等等。

三. Interceptors

okhttp-extension框架带有很多常用的拦截器

3.1 CurlLoggingInterceptor

将网络请求转换成 curl 命令的拦截器，便于后端同学调试排查问题。

以下面的代码为例：

    httpClient.get{

        url {
            url = "/response-headers-queries"

            "param1" to "value1"
            "param2" to "value2"
        }

        header {
            "key1" to "value1"
            "key2" to "value2"
        }
    }.use {
        println(it)
    }

添加了 CurlLoggingInterceptor 之后，打印结果如下：

curl:
╔══════════════════════════════════════════════════════════════════════════════════════════════════
║ curl -X GET -H "key1: value1" -H "key2: value2" "http://localhost:8080/response-headers-queries?param1=value1&param2=value2"
╚══════════════════════════════════════════════════════════════════════════════════════════════════

CurlLoggingInterceptor 默认使用 println 函数打印，可以使用相应的日志框架进行替换。

3.2 SigningInterceptor

请求签名的拦截器，支持对 query 参数进行签名。

const val TIME_STAMP = "timestamp"
const val NONCE = "nonce"
const val SIGN = "sign"

private val extraMap:MutableMap<String,String> = mutableMapOf<String,String>().apply {
    this[TIME_STAMP] = System.currentTimeMillis().toString()
    this[NONCE]  = UUID.randomUUID().toString()
}

private val signingInterceptor = SigningInterceptor(SIGN, extraMap, signer = {
    val paramMap = TreeMap<String, String>()
    val url = this.url

    for (name in url.queryParameterNames) {
        val value = url.queryParameterValues(name)[0]?:""
        paramMap[name] = value
    }

    //增加公共参数
    paramMap[TIME_STAMP] = extraMap[TIME_STAMP].toString()
    paramMap[NONCE]  = extraMap[NONCE].toString()

    //所有参数自然排序后拼接
    var paramsStr = join("",paramMap.entries
        .filter { it.key!= SIGN }
        .map { entry -> String.format("%s", entry.value) })

    //生成签名
    sha256HMAC(updateAppSecret,paramsStr)
})

3.3 TraceIdInterceptor

需要实现TraceIdProvider接口

interface TraceIdProvider {

    fun getTraceId():String
}

TraceIdInterceptor 会将 traceId 放入 http header 中。

3.4 OAuth2Interceptor

需要实现OAuth2Provider接口

interface OAuth2Provider {

    fun getOauthToken():String

    /**
     * 刷新token
     * @return String?
     */
    fun refreshToken(): String?
}

OAuth2Interceptor 会将 token 放入 http header 中，如果 token 过期，会调用 refreshToken() 方法进行刷新 token。

3.5 JWTInterceptor

需要实现JWTProvider接口

interface JWTProvider {

    fun getJWTToken():String

    /**
     * 刷新token
     * @return String?
     */
    fun refreshToken(): String?
}

JWTInterceptor 会将 token 放入 http header 中，如果 token 过期，会调用 refreshToken() 方法进行刷新 token。

3.6 LoggingInterceptor

可以使用我开发的okhttp-logging-interceptor将 http request、response 的数据格式化的输出。

四. Coroutines

Coroutines 是 Kotlin 的特性，我们使用okhttp-extension也可以很好地利用 Coroutines。

4.1 Coroutines

例如，最基本的使用

   "https://baidu.com".asyncGet()
       .await()
       .use {
           println(it)
       }

亦或者

        httpClient.asyncGet{

            url{
                url = "/response-headers-queries"

                "param1" to "value1"
                "param2" to "value2"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }
        }.await().use {
            println(it)
        }

以及

        httpClient.asyncPost{

            url {
                url = "/response-body"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }

            body("application/json") {
                json {
                    "key1" to "value1"
                    "key2" to "value2"
                    "key3" to "value3"
                }
            }
        }.await().use{
            println(it)
        }

asyncGet\asyncPost\asyncPut\asyncDelete\asyncHead\asyncPatch 函数在coroutines模块中，都是 HttpClient 的扩展函数，会返回Deferred<Response>对象。

同样，他们也是基于 DSL 的。

4.2 Flow

coroutines模块也提供了 flowGet\flowPost\flowPut\flowDelete\flowHead\flowPatch 函数，也是 HttpClient 的扩展函数，会返回Flow<Response>对象。

例如：

        httpClient.flowGet{

            url{
                url = "/response-headers-queries"

                "param1" to "value1"
                "param2" to "value2"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }
        }.collect {
            println(it)
        }

或者

        httpClient.flowPost{

            url {
                url = "/response-body"
            }

            header {
                "key1" to "value1"
                "key2" to "value2"
            }

            body("application/json") {
                json {
                    "key1" to "value1"
                    "key2" to "value2"
                    "key3" to "value3"
                }
            }
        }.collect{
            println(it)
        }

五. WebSocket

OkHttp 本身支持 WebSocket ，因此okhttp-extension对 WebSocket 做了一些增强，包括重连、连接状态的监听等。

5.1 Reconnect

在实际的应用场景中，WebSocket 的断线是经常发生的。例如：网络发生切换、服务器负载过高无法响应等都可能是 WebSocket 的断线的原因。

客户端一旦感知到长连接不可用，就应该发起重连。okhttp-extension的 ReconnectWebSocketWrapper 类是基于 OkHttp 的 WebSocket 实现的包装类，具有自动重新连接的功能。

在使用该包装类时，可以传入自己实现的 WebSocketListener 来监听 WebSocket 各个状态以及对消息的接收，该类也支持对 WebSocket 连接状态变化的监听、支持设置重连的次数和间隔。

例如：

    // 支持重试的 WebSocket 客户端
    ws = httpClient.websocket("http://127.0.0.1:9876/ws",listener = object : WebSocketListener() {

        override fun onOpen(webSocket: WebSocket, response: Response) {
            logger.info("connection opened...")

            websocket = webSocket

            disposable = Observable.interval(0, 15000,TimeUnit.MILLISECONDS) // 每隔 15 秒发一次业务上的心跳
                .subscribe({
                    heartbeat()
                }, {

                })
        }

        override fun onMessage(webSocket: WebSocket, text: String) {
            logger.info("received instruction: $text")
        }

        override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
        }

        override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
            logger.info("connection closing: $code, $reason")

            websocket = null

            disposable?.takeIf { !it.isDisposed }?.let {
                it.dispose()
            }
        }

        override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
            logger.error("connection closed: $code, $reason")

            websocket = null

            disposable?.takeIf { !it.isDisposed }?.let {
                it.dispose()
            }
        }

        override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
            logger.error("websocket connection error")

            websocket = null

            disposable?.takeIf { !it.isDisposed }?.let {
                it.dispose()
            }
        }
    },wsConfig = WSConfig())

5.2 onConnectStatusChangeListener

ReconnectWebSocketWrapper 支持对 WebSocket 连接状态的监听，只要实现onConnectStatusChangeListener即可。

    ws?.onConnectStatusChangeListener = {
        logger.info("${it.name}")
        status = it
    }

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2021/11/22 ，如有侵权请联系 cloudcommunity@tencent.com 删除

网站

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度