依赖
implementation("com.squareup.okhttp3:okhttp:3.14.2")
简单使用例子
final OkHttpClient client = new OkHttpClient();
final Request request = new Request.Builder()
.url("http://www.kuaidi100.com/query?type=yuantong&postid=11111111111")
.get()
.build();
//异步请求
Call asyncCall = client.newCall(request);
asyncCall.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
}
@Override
public void onResponse(Call call, Response response) throws IOException {
Log.e(TAG, "get响应体: " + response.body().string());
Log.d(TAG,System.currentTimeMillis()+""+Thread.currentThread().getName());
}
});
//同步请求
new Thread(new Runnable() {
@Override
public void run() {
try {
Response response = null;
response = client.newCall(request).execute();//得到Response 对象
if (response.isSuccessful()) {
Log.d(TAG,"response.code()=="+response.code());
Log.d(TAG,"response.message()=="+response.message());
Log.d(TAG,"res=="+response.body().string());
//此时的代码执行在子线程，修改UI的操作请使用handler跳转到UI线程。
}
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
Okhttp基本使用步骤如下:
- 创建 OkHttpClient 实例
- 创建 Request 实例
- 通过 OkHttpClient 和 Request 封装成一个请求 Call
- 同步请求调用 Call.execute()，异步请求调用 Call.enqueue()
源码分析
new OkHttpClient ()和new Request.Builder()比较简单，跳过这个步骤，从client.newCall(request)开始分析
/**
* Prepares the {@code request} to be executed at some point in the future.
*/
@Override public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
Call是一个接口，newCall()方法返回的对象类型是Call的实现类RealCall.也就是说同步执行请求的execute()和异步执行请求的enqueue()实现都在RealCall内.
execute()
//RealCall.java
@Override public Response execute() throws IOException {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
transmitter.timeoutEnter();
transmitter.callStart();
try {
client.dispatcher().executed(this);
//向服务器发送 request 并获得服务器的 response
return getResponseWithInterceptorChain();
} finally {
client.dispatcher().finished(this);
}
}
//RealCall.java
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
//创建一个拦截器链
List<Interceptor> interceptors = new ArrayList<>();
//应用拦截器
interceptors.addAll(client.interceptors());
interceptors.add(new RetryAndFollowUpInterceptor(client));
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, transmitter, null, 0,
originalRequest, this, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
boolean calledNoMoreExchanges = false;
try {
Response response = chain.proceed(originalRequest);
if (transmitter.isCanceled()) {
closeQuietly(response);
throw new IOException("Canceled");
}
return response;
} catch (IOException e) {
calledNoMoreExchanges = true;
throw transmitter.noMoreExchanges(e);
} finally {
if (!calledNoMoreExchanges) {
transmitter.noMoreExchanges(null);
}
}
}
这里引入了拦截器链，Request 需要通过拦截器链接收相应的处理最后才会发送到服务器并获取服务器返回的响应。
拦截器链的添加顺序与拦截器执行顺序一致:
-
应用拦截器:开发者添加的拦截器 -
retryAndFollowUpInterceptor:负责失败重连操作，以及重定向，如果 call 被取消会抛出 IOException -
BridgeInterceptor:作为网络层与应用层之间的桥梁，把(应用层请求)user request转化为(网络层请求)network request，然后向服务器发送network request，得到(网络层响应)network reseponse后转化为(应用层响应) user response -
CacheInterceptor:处理缓存中的 requests 以及把 responses 写到缓存 -
ConnectInterceptor:负责与目标服务器建立连接 -
网络拦截器:开发者添加的拦截器 -
CallServerInterceptor:拦截器链的最后一个拦截器，负责向服务器发送请求和从服务器获取响应数据
再来看看chain.proceed(originalRequest)的实现原理
//RealInterceptorChain.java
@Override public Response proceed(Request request) throws IOException {
return proceed(request, transmitter, exchange);
}
public Response proceed(Request request, Transmitter transmitter, @Nullable Exchange exchange)
throws IOException {
...
// Call the next interceptor in the chain.
//创建执行目标为下一个拦截器的RealInterceptorChain对象
RealInterceptorChain next = new RealInterceptorChain(interceptors, transmitter, exchange,
index + 1, request, call, connectTimeout, readTimeout, writeTimeout);
//获取当前目标拦截器
Interceptor interceptor = interceptors.get(index);
//调用当前拦截器的intercept()并传入下一个拦截器链
Response response = interceptor.intercept(next);
...
// Confirm that the intercepted response isn't null.
if (response == null) {
throw new NullPointerException("interceptor " + interceptor + " returned null");
}
if (response.body() == null) {
throw new IllegalStateException(
"interceptor " + interceptor + " returned a response with no body");
}
return response;
}
拦截器链的思路和工作原理：
- 把根据功能细分为多个拦截器。
- 按照拦截顺序组成一个拦截器链。
- Reuqest 拦截器链中传递(可被修改)并在每个拦截器中调用
Interceptor.intercept()完成各自的功能。 - 最后返回从服务器获取的响应
Response。
问题：拦截器链如何而实现按序调用Interceptor.intercept()？
主要涉及两个方法
/**
* Interceptor
* chain : 下一个拦截器
*/
Response intercept(Chain chain) throws IOException;
/**
* Interceptor.Chain
* request : 我们传的 request ,可以被拦截器修改，给最后的 CallServerInterceptor 向服务器发送请求
*/
Response proceed(Request request) throws IOException;
1.chain.proceed(request)会创建一个目标拦截器为下一个拦截器的拦截器链InterceptorChain对象 next ，然后调用当前目标拦截器的intercept(next)。
2.除了最后一个拦截器，其余拦截器的都在intercept()内直接或间接调用chain.proceed(request)重新回到上一步，然后重复刚才的步骤。这样就达到遍历拦截器链且按照顺序执行拦截器的具体操作的目的。
3.getResponseWithInterceptorChain()返回的是服务器的响应 Response，所以 CallServerInterceptor 必须在拦截器链的最后一个，因为它是真正向服务器发送请求和获取响应的拦截器。
enqueue()
@Override public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
transmitter.callStart();
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
public final class Dispatcher {
//同步请求和异步请求之和最大值
private int maxRequests = 64;
//同一个host请求数最大值
private int maxRequestsPerHost = 5;
private @Nullable Runnable idleCallback;
/** Executes calls. Created lazily. */
//用于执行请求的线程池，且是懒加载的
private @Nullable ExecutorService executorService;
/** Ready async calls in the order they'll be run. */
//等待被执行的异步请求
private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
/** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
//正在执行的异步请求
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
/** Running synchronous calls. Includes canceled calls that haven't finished yet. */
//正在执行的同步请求，包括已经被取消但未完成的请求
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();
public Dispatcher(ExecutorService executorService) {
this.executorService = executorService;
}
public Dispatcher() {
}
public synchronized ExecutorService executorService() {
if (executorService == null) {
//核心线程数为0，最大线程数为Integer.MAX_VALUE,空闲线程最多存活60s
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
...
void enqueue(AsyncCall call) {
synchronized (this) {
readyAsyncCalls.add(call);
// Mutate the AsyncCall so that it shares the AtomicInteger of an existing running call to
// the same host.
if (!call.get().forWebSocket) {
AsyncCall existingCall = findExistingCallWithHost(call.host());
if (existingCall != null) call.reuseCallsPerHostFrom(existingCall);
}
}
promoteAndExecute();
}
...
private boolean promoteAndExecute() {
assert (!Thread.holdsLock(this));
List<AsyncCall> executableCalls = new ArrayList<>();
boolean isRunning;
synchronized (this) {
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall asyncCall = i.next();
if (runningAsyncCalls.size() >= maxRequests) break; // Max capacity.
if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue; // Host max capacity.
i.remove();
asyncCall.callsPerHost().incrementAndGet();
executableCalls.add(asyncCall);
runningAsyncCalls.add(asyncCall);
}
isRunning = runningCallsCount() > 0;
}
for (int i = 0, size = executableCalls.size(); i < size; i++) {
AsyncCall asyncCall = executableCalls.get(i);
asyncCall.executeOn(executorService());
}
return isRunning;
}
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
...
}
Dispatcher 是任务调度器，内部建立了一个线程池 ExecutorService ，而且维护了三个集合:
- readyAsyncCalls : 等待被执行的异步请求集合
- runningAsyncCalls : 正在执行的异步请求集合，包括已经被取消但未完成的请求
- runningSyncCalls : 正在执行的同步请求集合，包括已经被取消但未完成的请求
所有异步请求都交由线程池 ExecutorService 来执行。
线程池其实是 ThreadPoolExecutor ，且核心线程数为 0 、最大线程数为Integer.MAX_VALUE、空闲线程存活最大时间为60秒，即所有线程执行完之后空闲60秒就会被销毁，而且存在线程过多导致内存溢出问题等问题，但是在 Dispatcher 的调度下是不会发生线程过多情况的，因为 Dispatcher 限制了正在执行的请求数(同步和异步之和)最大为64，同一个host下请求同时存在数最大值为 5 。
AsyncCall是一个实现了Runnable接口的类，通过线程池调用最终会执行AsyncCall的execute()方法
final class AsyncCall extends NamedRunnable {
...
@Override protected void execute() {
boolean signalledCallback = false;
transmitter.timeoutEnter();
try {
//通过拦截器链得到从服务器获取的响应 Response
Response response = getResponseWithInterceptorChain();
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 {
// 最后要通知 dispatcher 标记该任务已完成
client.dispatcher().finished(this);
}
}
}
AsyncCall的execute()逻辑很简单，调用getResponseWithInterceptorChain()方法得到Response对象，判断回调onResponse还是onFailure，因为是在线程池中调用的，所以enqueue()方法的回调是在子线程中被调用的，最后还要调用finished()通知Dispatcher该任务已经完成。getResponseWithInterceptorChain()方法在上面已经介绍过了
/** Used by {@code AsyncCall#run} to signal completion. */
void finished(AsyncCall call) {
call.callsPerHost().decrementAndGet();
finished(runningAsyncCalls, call);
}
/** Used by {@code Call#execute} to signal completion. */
void finished(RealCall call) {
finished(runningSyncCalls, call);
}
private <T> void finished(Deque<T> calls, T call) {
Runnable idleCallback;
synchronized (this) {
if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
idleCallback = this.idleCallback;
}
boolean isRunning = promoteAndExecute();
if (!isRunning && idleCallback != null) {
idleCallback.run();
}
}
同步请求和异步请求执行后最终都会调用 dispatcher.finish()
/**
* Promotes eligible calls from {@link #readyAsyncCalls} to {@link #runningAsyncCalls} and runs
* them on the executor service. Must not be called with synchronization because executing calls
* can call into user code.
*
* @return true if the dispatcher is currently running calls.
*/
private boolean promoteAndExecute() {
assert (!Thread.holdsLock(this));
List<AsyncCall> executableCalls = new ArrayList<>();
boolean isRunning;
synchronized (this) {
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall asyncCall = i.next();
if (runningAsyncCalls.size() >= maxRequests) break; // Max capacity.
if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue; // Host max capacity.
i.remove();
asyncCall.callsPerHost().incrementAndGet();
executableCalls.add(asyncCall);
runningAsyncCalls.add(asyncCall);
}
isRunning = runningCallsCount() > 0;
}
for (int i = 0, size = executableCalls.size(); i < size; i++) {
AsyncCall asyncCall = executableCalls.get(i);
asyncCall.executeOn(executorService());
}
return isRunning;
}
promoteAndExecute起到一个调度作用，将readyAsyncCalls中的任务放到runningAsyncCalls中去，并判断runningAsyncCalls是否超过允许最大值，满足条件则将call交给线程池执行。
手写okhttp架构参考代码：
https://github.com/games2sven/okhttp
附上okhttp的时序图
okhttp 时序图.jpg
