网络编程之——他山之石OkHttp

早在Android4.4的时代中，谷歌就已经把OkHttp框架引入了进来，据说6.0就要将apache的Http包替代掉，这足以说明这个框架是非常优秀的。本文将对其源码进行简要的剖析。

Android源码中和目前Github上的代码结构和细节存在一点差异，主要是Android将其中对Android的支持和对Httpurlconnection的扩展实现摘出来了并整合了一下。并且目前Android上的Httpurlconnection实现就是使用的OkHttp的HttpurlconnectionImpl，所以，OKHttp在这里也经常被当做一种更底层的网络访问实现来使用（并且它支持谷歌的Spdy协议）。

先给大家看一张类图，也就是OkHttp的主要类的示意图：

类还是蛮多的，主要有这么几个：

OkHttpClient：这是整个框架的入口，客户端编码直接使用它来进行网络请求以及响应处理。

Job：这是所有经过OkHttpClient提交的网络访问任务的封装。

Diapatcher：调度员顾名思义就是负责网络任务的执行调度，这里面是有一个线程池的，但是Client提交的job有可能不通过它而直接执行。

HttpEngine：引擎，负责的每个job的实际执行，提交请求，处理响应。

Request、Response：请求以及响应的封装。

Connection：抽象出来的网络连接。

Transport：负责数据的传输。

Sink和Source：抽象出来数据输入和输出通道。

OkResponseCache：网络请求的响应缓存。

DiskLruCache：缓存管理模块。

HttpURLConnectionImpl：框架中对HttpURLConnection的实现。

URL不属于该框架，这里列出来只是告诉大家URL.openURLConnection()为什么会调到HttpURLConnectionImpl这个实现中来。原因就是Android的URL类中加载了com.android.okhttp.HttpHandler类。

这些类中的主要方法在下面分析整个流程时再详细叙述。上面这些代码是目前在Android5.1的源码external\okhttp中的，与当前https://github.com/square/okhttp有不小的差别，但也只是类的封装不太一样，职责进行了个别变化，下图列出来的就是与前文所述架构不一致的地方。

下面我们通过一个简单的栗子来看看客户端如何使用OkHttpClient以及一次网络请求的具体流程是怎样。

Request request =new Request.Builder().url("").build();

new OkHttpClient().newCall(request).enqueue(new Callback() {

@Override

publicvoid onResponse(Responsearg0)throwsIOException {

//TODO Auto-generated method stub

}

@Override

publicvoid onFailure(Request arg0,IOException arg1) {

//TODO Auto-generated method stub

}

});

客户端的调用是如此的简单，这个简单是因为框架为你做了很多的事情，下面就针对这个简单的调用场景来分析一下框架工作的流程。下面是时序图：

时序图只是展示这个过程的一些主要流程，有些方法需要拿出来细细的说一下，前面关于请求以及线程池处理的部分比较清晰，那就从Job的getResponse()说起。

Response getResponse() throws IOException {

Response redirectedBy = null;

// Copy body metadata to the appropriaterequest headers.

Request.Body body = request.body();

//下面处理请求正文，包括填充一些header

if (body != null) {

MediaType contentType =body.contentType();

if (contentType == null) throw newIllegalStateException("contentType == null");

Request.Builder requestBuilder =request.newBuilder();

requestBuilder.header("Content-Type", contentType.toString());

long contentLength =body.contentLength();

if (contentLength != -1) {

requestBuilder.header("Content-Length",Long.toString(contentLength));

requestBuilder.removeHeader("Transfer-Encoding");

} else {

requestBuilder.header("Transfer-Encoding","chunked");

requestBuilder.removeHeader("Content-Length");

}

request = requestBuilder.build();

}

// Create the initial HTTP engine. Retries andredirects need new engine for each attempt.

//HttpEngine名字也起的很好就是http引擎，来完成客户端请求发送以及获取响应

engine = new HttpEngine(client, request, false, null,null, null, null);

//下面是一个循环，不断的请求直到成功（主要考虑了重定向）、取消或者IO异常

while (true) {

if(canceled) return null;

try {

engine.sendRequest();

//之前我一直以为下面这一块是往服务器写请求体，但是看到后面发现在readResponse中才是，这里只是往OKBuffer里面填充了而已，这个大的buffer一直在提高socket的流操作效率。

if (body != null) {

BufferedSink sink =Okio.buffer(engine.getRequestBody());

body.writeTo(sink);

sink.flush();

}

engine.readResponse();

//到这是正常的一次请求。

} catch (IOException e) {

HttpEngine retryEngine =engine.recover(e);

if (retryEngine != null) {

engine = retryEngine;

continue;

}

// Give up; recovery is not possible.

throw e;

}

//处理响应，确认是否需要重定向

Response response = engine.getResponse();

Request redirect = processResponse(engine,response);

if (redirect == null) {

//不需要重定向直接释放连接，返回响应数据

engine.releaseConnection();

return response.newBuilder()

.body(newRealResponseBody(response, engine.getResponseBody()))

.priorResponse(redirectedBy)

.build();

}

//判断重定向和原请求是否为同一个连接，包括比较host、port以及Protocol，如果不是就可以释放连接

if (!sameConnection(request, redirect)) {

engine.releaseConnection();

}

//准备下一次重定向访问

Connection connection = engine.close();

redirectedBy =response.newBuilder().priorResponse(redirectedBy).build(); // Chained.

request = redirect;

engine = new HttpEngine(client, request,false, connection, null, null, null);

}

里面最重要的两个方法：engine.sendRequest()，engine.readResponse()，分别是发生请求和读取响应，但是其实真正往socket里面写数据和读数据都在readResponse()里面。

public final void sendRequest() throwsIOException {

if (responseSource != null) return; //Already sent.

if (transport != null) throw newIllegalStateException();

//把用户请求封装一下，添加一些必须要的header等内容

Request request = networkRequest(userRequest);

//下面都是缓存处理的一些逻辑，可以先跳过，后面单独说缓存的实现。这里只看不使用缓存的情况，最终会生成一个networkRequest就是实际上的网络请求。

OkResponseCache responseCache =client.getOkResponseCache();

Response cacheCandidate = responseCache !=null

?responseCache.get(request)

: null;

long now = System.currentTimeMillis();

CacheStrategy cacheStrategy = newCacheStrategy.Factory(now, request, cacheCandidate).get();

responseSource = cacheStrategy.source;

networkRequest = cacheStrategy.networkRequest;

cacheResponse =cacheStrategy.cacheResponse;

if (responseCache != null) {

responseCache.trackResponse(responseSource);

}

if (cacheCandidate != null

&& (responseSource ==ResponseSource.NONE || cacheResponse == null)) {

closeQuietly(cacheCandidate.body()); //The cache candidate wasn't applicable. Close it.

}

if (networkRequest != null) {

// Open a connection unless we inheritedone from a redirect.首次使用创建connection，后面重定向的情况过来直接复用的。

if (connection == null) {

connect(networkRequest);

}

// Blow up if we aren't the current ownerof the connection.

if (connection.getOwner() != this&& !connection.isSpdy()) throw new AssertionError();

transport = (Transport)connection.newTransport(this);

// Create a request body if we don't haveone already. We'll already have

// one if we're retrying a failed POST.

if (hasRequestBody() &&requestBodyOut == null) {

//准备请求的正文

requestBodyOut =transport.createRequestBody(request);

}

} else {

// We're using a cached response. Recyclea connection we may have inherited from a redirect.

if (connection != null) {

client.getConnectionPool().recycle(connection);

connection = null;

}

// No need for the network! Promote thecached response immediately.

//如果不需要重新访问网络即使用缓存则直接根据cacheresponse生成

this.userResponse = cacheResponse.newBuilder()

.request(userRequest)

.priorResponse(stripBody(priorResponse))

.cacheResponse(stripBody(cacheResponse))

.build();

if (userResponse.body() != null) {

initContentStream(userResponse.body().source());

}

//然后是读取响应，实际的网络数据处理都在这里进行的

public final voidreadResponse() throws IOException {

if (userResponse != null) {

return; // Already ready.

}

if (networkRequest == null &&cacheResponse == null) {

throw newIllegalStateException("call sendRequest() first!");

}

if (networkRequest == null) {

return; // No network response to read.

}

// Flush the request body if there's dataoutstanding.

if (bufferedRequestBody != null &&bufferedRequestBody.buffer().size() > 0) {

bufferedRequestBody.flush();

}

if (sentRequestMillis == -1) {

if(OkHeaders.contentLength(networkRequest) == -1

&& requestBodyOut instanceofRetryableSink) {

// We might not learn theContent-Length until the request body has been buffered.

long contentLength = ((RetryableSink)requestBodyOut).contentLength();

networkRequest =networkRequest.newBuilder()

.header("Content-Length",Long.toString(contentLength))

.build();

}

//写入请求头

transport.writeRequestHeaders(networkRequest);

}

if (requestBodyOut != null) {

if (bufferedRequestBody != null) {

// This also closes the wrappedrequestBodyOut.

bufferedRequestBody.close();

} else {

requestBodyOut.close();

}

if (requestBodyOut instanceofRetryableSink) {

//写入请求体

transport.writeRequestBody((RetryableSink)requestBodyOut);

}

//写入数据的提交

transport.flushRequest();

//等待处理响应，先读取响应头

networkResponse =transport.readResponseHeaders()

.request(networkRequest)

.handshake(connection.getHandshake())

.header(OkHeaders.SENT_MILLIS,Long.toString(sentRequestMillis))

.header(OkHeaders.RECEIVED_MILLIS,Long.toString(System.currentTimeMillis()))

.setResponseSource(responseSource)

.build();

connection.setHttpMinorVersion(networkResponse.httpMinorVersion());

receiveHeaders(networkResponse.headers());

if (responseSource ==ResponseSource.CONDITIONAL_CACHE) {

if(cacheResponse.validate(networkResponse)) {

userResponse =cacheResponse.newBuilder()

.request(userRequest)

.priorResponse(stripBody(priorResponse))

.headers(combine(cacheResponse.headers(), networkResponse.headers()))

.cacheResponse(stripBody(cacheResponse))

.networkResponse(stripBody(networkResponse))

.build();

transport.emptyTransferStream();

releaseConnection();

// Update the cache after combiningheaders but before stripping the

// Content-Encoding header (asperformed by initContentStream()).

OkResponseCache responseCache =client.getOkResponseCache();

responseCache.trackConditionalCacheHit();

responseCache.update(cacheResponse,stripBody(userResponse));

if (cacheResponse.body() != null) {

initContentStream(cacheResponse.body().source());

}

return;

} else {

closeQuietly(cacheResponse.body());

}

userResponse = networkResponse.newBuilder()

.request(userRequest)

.priorResponse(stripBody(priorResponse))

.cacheResponse(stripBody(cacheResponse))

.networkResponse(stripBody(networkResponse))

.build();

if (!hasResponseBody()) {

// Don't call initContentStream() whenthe response doesn't have any content.

responseTransferSource =transport.getTransferStream(storeRequest);

responseBody = responseTransferSource;

return;

}

maybeCache();

//这是最关键的，这里会根据网络响应生成一个Source用于读取响应正文

initContentStream(transport.getTransferStream(storeRequest));

}

接下来看一下getTransferStream()方法里面做了什么事情。其实就是new了一个Source。

@Override public SourcegetTransferStream(CacheRequest cacheRequest) throws IOException {

if (!httpEngine.hasResponseBody()) {

returnhttpConnection.newFixedLengthSource(cacheRequest, 0);

}

if("chunked".equalsIgnoreCase(httpEngine.getResponse().header("Transfer-Encoding"))){

returnhttpConnection.newChunkedSource(cacheRequest, httpEngine);

}

long contentLength =OkHeaders.contentLength(httpEngine.getResponse());

if (contentLength != -1) {

returnhttpConnection.newFixedLengthSource(cacheRequest, contentLength);

}

// Wrap the input stream from theconnection (rather than just returning

// "socketIn" directly here), sothat we can control its use after the

// reference escapes.

returnhttpConnection.newUnknownLengthSource(cacheRequest);

}

//下面就是在有响应正文的情况下使用的Source，read就是当客户端从网络socket读取数据的接口，通过HttpConnection中的source来操作的。

private class FixedLengthSource extendsAbstractSource implements Source {

private long bytesRemaining;

public FixedLengthSource(CacheRequestcacheRequest, long length) throws IOException {

super(cacheRequest);

bytesRemaining = length;

if (bytesRemaining == 0) {

endOfInput(true);

}

@Override public long read(OkBuffer sink,long byteCount)

throws IOException {

if (byteCount < 0) throw newIllegalArgumentException("byteCount < 0: " + byteCount);

if (closed) throw newIllegalStateException("closed");

if (bytesRemaining == 0) return -1;

long read = source.read(sink,Math.min(bytesRemaining, byteCount));

if (read == -1) {

unexpectedEndOfInput(); // the serverdidn't supply the promised content length

throw newProtocolException("unexpected end of stream");

}

bytesRemaining -= read;

cacheWrite(sink, read);

if (bytesRemaining == 0) {

endOfInput(true);

}

return read;

}

@Override public Source deadline(Deadlinedeadline) {

source.deadline(deadline);

return this;

}

@Override public void close() throwsIOException {

if (closed) return;

if (bytesRemaining != 0 &&!discard(this, DISCARD_STREAM_TIMEOUT_MILLIS)) {

unexpectedEndOfInput();

}

closed = true;

}

整个流程就说到这里，里面有几个好的设计点，比如：Cache、连接池和Buffer这三个设计。下面分别说一下。

Cache是指响应缓存，是一个OkResponseCache接口，主要实现是HttpResponseCache，作用不言而喻就是做为本地的响应缓存，以提高请求的响应速度。采用的算法是LRU最近最少使用（DiskLruCache），存储在文件中。另外实现了一个ResponseCacheAdapter用来作为java.net.ResponseCache和OkResponseCache的适配器。

整个工作流程是这样的：

1、OkHttpClient中通过接口setResponseCache设置缓存。

2、设置请求的headers控制是否使用缓存，headers会被解析成一个CacheControl。“Cache-Control”。

3、请求的处理过程中先读取对应请求的缓存并根据策略判断是否需要使用缓存（这个策略在CacheStrategy中定义，具体可以参见getCandidate方法），如果可以使用缓存则直接返回，不使用则继续进行网络访问。

4、网络访问完成之后读取响应时会判断当前响应是否需要缓存起来，具体方法是maybeCache，具体的策略也是在CacheStrategy中定义具体方法参见isCacheable()。

ConnectionPool连接池，通过掌握socket的创建与关闭来重用connection。用两种清除模式：NORMAL每次有新的连接加入时都会清除掉所有不在活动状态的并且不在保持间隔内的空闲连接；DRAINED是以固定间隔来执行清理不活动的空闲连接。

在发送请求sendRequest时，如果判断需要进行网络访问此时则会根据特定的一个路由信息Route（地址端口代理等等）来创建或者复用一个connection。
之后分两种情况：如果是Spdy直接将这个连接share到连接池中。如果是http则会在网络访问结束之后releaseConnection，当然这里说是release，到底是关闭还是保留进入pool中还是需要判断请求和响应的headers是否配置了Connection：close。

OkBuffer读写缓冲，可以在写请求以及读响应的很多地方看到它的身影。里面其实就是一个用链表实现的SegmentPool，最大为64 KiB。涉及到byte操作时都会先拷贝到Segment其大小限制为2KiB，然后再跟其他流进行输入输出操作。这个环节也是上面时序图与socket最后交互部分流程不清晰的原因，下面就仔细分析一下。

socket的流读写抽象出了两个管道一个是source另一个是sink两者都是相对于客户端来说的，分别对应的是InputStream和OutputStream。为了使用Buffer又在此基础上扩展出BufferedSource和BufferedSink。

举一个发送请求headers的栗子：大概的方法调用堆栈如下，关键的buffer起作用的地方在RealBufferedSink里面，就是这个buffer对象。

Write操作中会先将数据写到buffer里面然后才会从这个里面写到OutputStream里面去。

刚刚开始看的时候我就有个疑问，SegmentPool是一个单例的，那么多个线程多个Okbuffer对象都会对其进行操作，会不会造成数据的错乱呢。原来纯是我想多了，这个SegmentPool只是作为未使用的Segment的池，每个线程对应一个请求也就是一个HttpEngine对应一个Okbuffer，需要提交和读取的数据Segmen都是保存在这个Okbuffer对象里面的。SegmentPool只需要保证take和recycle操作时是线程安全的也就是不会取到重复的Segment对象。

更多相关文章

随机推荐