本文基于'com.squareup.okhttp3:okhttp:3.11.0'分析
okhttp github网址： https://github.com/square/okhttp
第一步基本使用
同步GET
new Thread(new Runnable() {
@Override
public void run() {
try {
String url = "https://api.github.com/";
OkHttpClient client = new OkHttpClient();
Request request = new Request.Builder().url(url).build();
okhttp3.Response response = client.newCall(request).execute();
if (response.isSuccessful()) {
Log.i(TAG, response.body().string());
} else {
Log.i(TAG, "okHttp is request error");
}
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
异步GET
try {
Log.i(TAG,"main thread id is "+Thread.currentThread().getId());
String url = "https://api.github.com/";
OkHttpClient client = new OkHttpClient();
Request request = new Request.Builder().url(url).build();
client.newCall(request).enqueue(new okhttp3.Callback() {
@Override
public void onFailure(okhttp3.Call call, IOException e) {
}
@Override
public void onResponse(okhttp3.Call call, okhttp3.Response response) throws IOException {
// 该回调是子线程，非主线程
Log.i(TAG,"callback thread id is "+Thread.currentThread().getId());
Log.i(TAG,response.body().string());
}
});
} catch (Exception e) {
e.printStackTrace();
}
POST
new Thread(new Runnable() {
@Override
public void run() {
try {
// 请求完整url：http://api.k780.com:88/?app=weather.future&weaid=1&&appkey=10003&sign=b59bc3ef6191eb9f747dd4e83c99f2a4&format=json
String url = "http://api.k780.com:88/";
OkHttpClient okHttpClient = new OkHttpClient();
RequestBody formBody = new FormBody.Builder().add("app", "weather.future")
.add("weaid", "1").add("appkey", "10003").add("sign",
"b59bc3ef6191eb9f747dd4e83c99f2a4").add("format", "json")
.build();
Request request = new Request.Builder().url(url).post(formBody).build();
okhttp3.Response response = okHttpClient.newCall(request).execute();
Log.i(TAG, response.body().string());
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
第二步 源码分析
先分析sync get请求
OkHttpClient client = new OkHttpClient();
public OkHttpClient() {
this(new Builder());
}
private OkHttpClient(Builder builder) {
this.dispatcher = builder.dispatcher;
this.proxy = builder.proxy;
this.protocols = builder.protocols;
this.connectionSpecs = builder.connectionSpecs;
this.interceptors = Util.immutableList(builder.interceptors);
this.networkInterceptors = Util.immutableList(builder.networkInterceptors);
this.eventListenerFactory = builder.eventListenerFactory;
this.proxySelector = builder.proxySelector;
this.cookieJar = builder.cookieJar;
this.cache = builder.cache;
this.internalCache = builder.internalCache;
this.socketFactory = builder.socketFactory;
...
}
//OkHttpClient$Builder
public Builder() {
dispatcher = new Dispatcher();
protocols = DEFAULT_PROTOCOLS;
connectionSpecs = DEFAULT_CONNECTION_SPECS;
eventListenerFactory = EventListener.factory(EventListener.NONE);
proxySelector = ProxySelector.getDefault();
cookieJar = CookieJar.NO_COOKIES;
socketFactory = SocketFactory.getDefault();
hostnameVerifier = OkHostnameVerifier.INSTANCE;
certificatePinner = CertificatePinner.DEFAULT;
proxyAuthenticator = Authenticator.NONE;
authenticator = Authenticator.NONE;
connectionPool = new ConnectionPool();
...
}
首先是一个建造者模式，新建默认的dispatcher socketFactory等
之后构建Request，也是request的建造者模式。
Request request = new Request.Builder().url(url).build();
//Request$Builder
public Builder() {
this.method = "GET";
this.headers = new Headers.Builder();
}
//.url(url)
public Builder url(String url) {
if (url == null) throw new NullPointerException("url == null");
// Silently replace web socket URLs with HTTP URLs.
if (url.regionMatches(true, 0, "ws:", 0, 3)) {
url = "http:" + url.substring(3);
} else if (url.regionMatches(true, 0, "wss:", 0, 4)) {
url = "https:" + url.substring(4);
}
return url(HttpUrl.get(url));
}
先是Request.Builder()对应的Request$Builder，创建GET方式请求。
之后是.url(url)，主要是HttpUrl.get(url)解析地址的合法性，构建并返回HttpUrl对象。之后通过.build()构建Request。
okhttp3.Response response = client.newCall(request).execute();
//OkHttpClient
public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
//RealCall
private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
this.client = client;
this.originalRequest = originalRequest;
this.forWebSocket = forWebSocket;
this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
}
//.execute()
public Response execute() throws IOException {
synchronized (this) {
//检测请求是否已经在执行
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
try {
//将此request分发，实则放到同步运行call队列中
client.dispatcher().executed(this);
Response result = getResponseWithInterceptorChain();
if (result == null) throw new IOException("Canceled");
return result;
} catch (IOException e) {
eventListener.callFailed(this, e);
throw e;
} finally {
//将此分发移除掉
client.dispatcher().finished(this);
}
}
//Dispatcher
/** Running synchronous calls. Includes canceled calls that haven't finished yet. */
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
//client.dispatcher().execute(this)
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
对于sync请求，调用execute()会将request分发(即放到同步运行call队列中)。
然后是getResponseWithInterceptorChain(false)调用拦截器责任链对请求处理
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
List<Interceptor> interceptors = new ArrayList<>();
interceptors.addAll(client.interceptors());
interceptors.add(retryAndFollowUpInterceptor);
interceptors.add(new BridgeInterceptor(client.cookieJar()));
interceptors.add(new CacheInterceptor(client.internalCache()));
interceptors.add(new ConnectInterceptor(client));
if (!forWebSocket) {
interceptors.addAll(client.networkInterceptors());
}
interceptors.add(new CallServerInterceptor(forWebSocket));
Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
originalRequest, this, eventListener, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
return chain.proceed(originalRequest);
}
//RealInterceptorChain#proceed()被调用
final class RealInterceptorChain implements Interceptor.Chain{
...
public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
RealConnection connection) throws IOException {
...
//执行index+1的intercept
RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
connection, index + 1, request, call, eventListener, connectTimeout, readTimeout,
writeTimeout);
Interceptor interceptor = interceptors.get(index);
Response response = interceptor.intercept(next);
...
}
}
先是添加Interceptor，具体有如下
retryAndFollowUpInterceptor：负责失败重试及重定向
BridgeInterceptor：负责根据用户请求建立网络请求，收到结果后将网络结果转换为用户结果
CacheInterceptor：负责从缓存中取结果或者网络请求后保存到缓存中
ConnectInterceptor：负责打开网络连接
networkInterceptors：用户配置的网络拦截器
CallServerInterceptor：负责具体网络请求的拦截器
然后获取第index个interceptor，执行interceptor.intercept(next)，并传入下一个interceptor，进入下一个interceptor会调用RealInterceptorChain#proceed()方法，然后在proceed()方法内获取第index+1个interceptor，并执行interceptor.intercept(next)，如此循环往复。
拦截器定义如下
public interface Interceptor {
Response intercept(Chain chain) throws IOException;
interface Chain {
Request request();
Response proceed(Request request) throws IOException;
...
}
下面看retryAndFollowUpInterceptor
public final class RetryAndFollowUpInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
RealInterceptorChain realChain = (RealInterceptorChain) chain;
try {
response = realChain.proceed(request, streamAllocation, null, null);
releaseConnection = false;
} catch (RouteException e) {
// The attempt to connect via a route failed. The request will not have been sent.
if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
throw e.getFirstConnectException();
}
releaseConnection = false;
continue;
}
...
try {
followUp = followUpRequest(response, streamAllocation.route());
} catch (IOException e) {
streamAllocation.release();
throw e;
}
...
}
}
如果有异常则调用recover重试，followUpRequest重定向
下面看BridgeInterceptor
public final class BridgeInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
Request userRequest = chain.request();
Request.Builder requestBuilder = userRequest.newBuilder();
...
if (userRequest.header("Host") == null) {
requestBuilder.header("Host", hostHeader(userRequest.url(), false));
}
...
Response networkResponse = chain.proceed(requestBuilder.build());
Response.Builder responseBuilder = networkResponse.newBuilder()
.request(userRequest);
return responseBuilder.build();
}
}
可以看到是根据用户请求创建服务器请求，拿到结果后根据服务器结果转换成用户结果。
下面看ConnectInterceptor
/** Opens a connection to the target server and proceeds to the next interceptor. */
public final class ConnectInterceptor implements Interceptor {
public final OkHttpClient client;
public ConnectInterceptor(OkHttpClient client) {
this.client = client;
}
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request();
StreamAllocation streamAllocation = realChain.streamAllocation();
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");
HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
RealConnection connection = streamAllocation.connection();
return realChain.proceed(request, streamAllocation, httpCodec, connection);
}
}
主要是创建一个httpCodec，它利用Okio对socket进行操作
然后是CacheInterceptor
public final class CacheInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
Request networkRequest = strategy.networkRequest;
Response cacheResponse = strategy.cacheResponse;
...
// If we don't need the network, we're done.
if (networkRequest == null) {
return cacheResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.build();
}
...
try {
networkResponse = chain.proceed(networkRequest);
} finally {
// If we're crashing on I/O or otherwise, don't leak the cache body.
if (networkResponse == null && cacheCandidate != null) {
closeQuietly(cacheCandidate.body());
}
}
...
Response response = networkResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.networkResponse(stripBody(networkResponse))
.build();
return response;
}
}
先是获取配置的缓存策略，然后看是从内存还是从网络获取，根据返回的结果建立Response.
最后是CallServerInterceptor
public final class CallServerInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
HttpCodec httpCodec = realChain.httpStream();
StreamAllocation streamAllocation = realChain.streamAllocation();
RealConnection connection = (RealConnection) realChain.connection();
...
realChain.eventListener().requestHeadersStart(realChain.call());
httpCodec.writeRequestHeaders(request);//发送RequestHeaders
realChain.eventListener().requestHeadersEnd(realChain.call(), request);
Response.Builder responseBuilder = null;
if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
// If there's a "Expect: 100-continue" header on the request, wait for a "HTTP/1.1 100
// Continue" response before transmitting the request body. If we don't get that, return
// what we did get (such as a 4xx response) without ever transmitting the request body.
if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
httpCodec.flushRequest();
realChain.eventListener().responseHeadersStart(realChain.call());
responseBuilder = httpCodec.readResponseHeaders(true);
}
if (responseBuilder == null) {
// Write the request body if the "Expect: 100-continue" expectation was met.
realChain.eventListener().requestBodyStart(realChain.call());
long contentLength = request.body().contentLength();
CountingSink requestBodyOut =
new CountingSink(httpCodec.createRequestBody(request, contentLength));
BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
request.body().writeTo(bufferedRequestBody);//通过Okio将请求发出去
bufferedRequestBody.close();
realChain.eventListener()
.requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);
} else if (!connection.isMultiplexed()) {
// If the "Expect: 100-continue" expectation wasn't met, prevent the HTTP/1 connection
// from being reused. Otherwise we're still obligated to transmit the request body to
// leave the connection in a consistent state.
streamAllocation.noNewStreams();
}
}
httpCodec.finishRequest();
if (responseBuilder == null) {
realChain.eventListener().responseHeadersStart(realChain.call());
responseBuilder = httpCodec.readResponseHeaders(false);
}
Response response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
int code = response.code();
if (code == 100) {
// server sent a 100-continue even though we did not request one.
// try again to read the actual response
responseBuilder = httpCodec.readResponseHeaders(false);
response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
code = response.code();
}
realChain.eventListener()
.responseHeadersEnd(realChain.call(), response);
if (forWebSocket && code == 101) {
// Connection is upgrading, but we need to ensure interceptors see a non-null response body.
response = response.newBuilder()
.body(Util.EMPTY_RESPONSE)
.build();
} else {
response = response.newBuilder()
.body(httpCodec.openResponseBody(response))
.build();
}
if ("close".equalsIgnoreCase(response.request().header("Connection"))
|| "close".equalsIgnoreCase(response.header("Connection"))) {
streamAllocation.noNewStreams();
}
if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
throw new ProtocolException(
"HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
}
return response;
}
}
主要是通过httpCodec发送请求，readResponseHeaders读取返回的头，httpCodec.openResponseBody(response)读取返回的内容，构建response返回。
接下来是Async get请求
client.newCall(request).enqueue(new okhttp3.Callback() {...})
//RealCall
public void enqueue(Callback responseCallback) {
enqueue(responseCallback, false);
}
void enqueue(Callback responseCallback, boolean forWebSocket) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
//将请求分发
client.dispatcher().enqueue(new AsyncCall(responseCallback, forWebSocket));
}
synchronized void enqueue(AsyncCall call) {
//如果正在执行的请求小于最大请求数并且在这个主机上请求数也小于一个主机最大请求书，那么把call放到runningAsyncCalls中，并且开启线程池执行请求
if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
runningAsyncCalls.add(call);
executorService().execute(call);
} else {
readyAsyncCalls.add(call);
}
}
/** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();//先进先出队列
public synchronized ExecutorService executorService() {
if (executorService == null) {
//最小并发数是0，60s就销毁所有线程
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
将请求放入线程池中准备执行。
AsyncCall继承自NamedRunnable，而NamedRunnable是一个Runnable，所以AsyncCall也是一个Runnable
public abstract class NamedRunnable implements Runnable {
protected final String name;
public NamedRunnable(String format, Object... args) {
this.name = Util.format(format, args);
}
@Override public final void run() {
String oldName = Thread.currentThread().getName();
Thread.currentThread().setName(name);
try {
execute();
} finally {
Thread.currentThread().setName(oldName);
}
}
protected abstract void execute();
}
线程池开始执行会调用AsyncCall的execute方法
//AsyncCall
protected void execute() {
boolean signalledCallback = false;
try {
//获取结果
Response response = getResponseWithInterceptorChain(forWebSocket);
if (canceled) {
signalledCallback = true;
responseCallback.onFailure(RealCall.this, new IOException("Canceled"));
} else {
signalledCallback = true;
responseCallback.onResponse(RealCall.this, response);
}
} catch (IOException e) {
if (signalledCallback) {
// Do not signal the callback twice!
Platform.get().log(INFO, "Callback failure for " + toLoggableString(), e);
} else {
responseCallback.onFailure(RealCall.this, e);
}
} finally {
client.dispatcher().finished(this);
}
}
先看线程池如何从runningAsyncCalls及readyAsyncCalls中循环取出call并执行，在这里主要是通过finally执行client.dispatcher().finished(this)取出来的
synchronized void finished(AsyncCall call) {
if (!runningAsyncCalls.remove(call)) throw new AssertionError("AsyncCall wasn't running!");
promoteCalls();
}
//将call从readyAsyncCalls中取出来放到runningAsyncCalls里，同时放如线程池中执行
private void promoteCalls() {
if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall call = i.next();
if (runningCallsForHost(call) < maxRequestsPerHost) {
i.remove();
runningAsyncCalls.add(call);
executorService().execute(call);
}
if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
}
}
之后getResponseWithInterceptorChain方法同sync get流程一样。
Okhttp使用了一个线程池来进行异步网络任务的真正执行，而对于任务的管理采用了任务队列的模型。Okhttp使用分发器Dispatcher来维护一个running的任务队列和一个ready的队列。如果当前并发任务数量小于64，就放入running队列中并且放入线程池中执行，而如果当前并发数量大于64就放入ready队列中。在每次有任务执行完成之后就在任务的finally块中调用分发器的finish函数，在ready队列中查看是否有空余任务，如果有就进行入队执行。Okhttp就是使用任务队列的模型来进行任务的执行和调度的。
网友评论