menu
more_vert

教你写Http框架(二)——三个样例带你深入理解AsyncTask

作者:dayu

private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors(); private static final int CORE_POOL_SIZE = CPU_COUNT + 1; private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1; private static final int KEEP_ALIVE = 1; private static final BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>(10); private static final ThreadFactory threadFactory = new ThreadFactory() { private final AtomicInteger count = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { return new Thread(r, "AsyncTask #" + count.getAndIncrement()); } }; private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor( CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE, TimeUnit.SECONDS, workQueue, threadFactory);

敏感的同学会发现。这个就是AsyncTask的线程池的源代码,的确。正常的需求,这段代码实例化出来的线程池基本都能够满足了。其它參数看命名都非常容易理解,这里主要讲一下workQueue。由于我们会不断提交任务给线程池运行,而线程池的线程数量是有限的,当全部核心线程都处于工作状态时。client再次提交的任务放在哪里呢?我这么一问你就懂了吧。


再讲一下java的FutureTask,我们知道正常情况下我们须要一个线程运行,提交的是一个Runnable。但有时候我们希望线程运行结束时带回一个处理完毕的数据。这个时候Runnable就无力了,这个时候就要看FutureTask了。大家有兴趣能够看一下它的源代码。事实上它也是继承自Runnable的,所以能够直接提交给线程来运行。

一般正常调用FutureTask的方法例如以下代码:

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class Test1
{
    public static void main(String[] args)
    {
        Test1 test = new Test1();
        test.test();
    }

    public void test()
    {
        FutureTask<String> fTask = new FutureTask<String>(new Callable<String>()
        {
            @Override
            public String call() throws Exception
            {
                System.out.println("calling");
                return "hello";
            }
        })
        {
            @Override
            protected void done()
            {
                try
                {
                    System.out.println("done " + get());
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
                catch (ExecutionException e)
                {
                    e.printStackTrace();
                }

                super.done();
            }
        };

        Executor executor = Executors.newSingleThreadExecutor();
        executor.execute(fTask);
    }


}

以上代码的运行结果为:
calling
done hello
所以,我们在线程结束时拿到了终于的线程处理结果。而AsyncTask在onPostExecute中给你结果的时候,就是这么干的。

第二个样例,我们来点干货。我们先写个AsyncTask的样例,跑起来并看下运行结果。先代码:

package com.amuro.activity;

import android.app.Activity;
import android.os.AsyncTask;
import android.os.Bundle;
import android.util.Log;
import android.view.View;

import com.amuro.R;

public class MainActivity extends Activity
{

    @Override
    protected void onCreate(Bundle savedInstanceState)
    {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main_layout);

        findViewById(R.id.bt).setOnClickListener(new View.OnClickListener()
        {
            @Override
            public void onClick(View v)
            {
               testAsync();
            }

        });
    }

    private void testAsync()
    {
        for(int i = 0; i < 10; i++)
        {
            final int j = i;
            AsyncTask<String, Integer, String> aTask =
                    new AsyncTask<String, Integer, String>()
            {
                @Override
                protected void onProgressUpdate(Integer... values)
                {
                    super.onProgressUpdate(values);
                }

                @Override
                protected String doInBackground(String... params)
                {
                    Log.e("amuro", Thread.currentThread().getName());
                    try
                    {
                        Thread.sleep(3000);
                    }
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    return params[0] + "done";
                }

                @Override
                protected void onPostExecute(String s)
                {
                    Log.e("amuro", "result: " + s + " " + j);
                }
            };
            aTask.execute("DoubleX");
        }
    }
}

看下运行结果:
03-13 11:23:47.950 22777-23081/com.amuro E/amuro: AsyncTask #1
03-13 11:23:50.955 22777-22777/com.amuro E/amuro: result: DoubleXdone 0
03-13 11:23:50.955 22777-23120/com.amuro E/amuro: AsyncTask #2
03-13 11:23:53.960 22777-22777/com.amuro E/amuro: result: DoubleXdone 1
03-13 11:23:53.960 22777-23195/com.amuro E/amuro: AsyncTask #3
03-13 11:23:56.965 22777-22777/com.amuro E/amuro: result: DoubleXdone 2
03-13 11:23:56.965 22777-23236/com.amuro E/amuro: AsyncTask #4
03-13 11:23:59.960 22777-22777/com.amuro E/amuro: result: DoubleXdone 3
03-13 11:23:59.965 22777-23277/com.amuro E/amuro: AsyncTask #5
03-13 11:24:02.965 22777-22777/com.amuro E/amuro: result: DoubleXdone 4
03-13 11:24:02.965 22777-23277/com.amuro E/amuro: AsyncTask #5
03-13 11:24:05.965 22777-22777/com.amuro E/amuro: result: DoubleXdone 5
03-13 11:24:05.970 22777-23277/com.amuro E/amuro: AsyncTask #5
03-13 11:24:08.975 22777-22777/com.amuro E/amuro: result: DoubleXdone 6
03-13 11:24:08.975 22777-23277/com.amuro E/amuro: AsyncTask #5
03-13 11:24:11.975 22777-22777/com.amuro E/amuro: result: DoubleXdone 7
03-13 11:24:11.975 22777-23277/com.amuro E/amuro: AsyncTask #5
03-13 11:24:14.980 22777-22777/com.amuro E/amuro: result: DoubleXdone 8
03-13 11:24:14.980 22777-23081/com.amuro E/amuro: AsyncTask #1
03-13 11:24:17.985 22777-22777/com.amuro E/amuro: result: DoubleXdone 9
能够看到,10个任务是顺序运行的,而且仅仅有5个线程在工作,好,
我们把AsyncTask刚才那个线程池和FutureTask结合起来,写一个简单的样例实现和它一模一样的功能。代码:

package com.amuro;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Test2
{

    public static void main(String[] args)
    {
        Test2 test = new Test2();
        test.test();
    }

    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = CPU_COUNT + 1;
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
    private static final int KEEP_ALIVE = 1;

    private static final BlockingQueue<Runnable> workQueue =
            new LinkedBlockingQueue<Runnable>(10);
    private static final ThreadFactory threadFactory = new ThreadFactory()
    {
        private final AtomicInteger count = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r)
        {
            return new Thread(r, "AsyncTask #" + count.getAndIncrement());
        }
    };
    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR =
            new ThreadPoolExecutor(
                    CORE_POOL_SIZE, 
                    MAXIMUM_POOL_SIZE, 
                    KEEP_ALIVE, 
                    TimeUnit.SECONDS, 
                    workQueue, 
                    threadFactory);

    private static volatile Executor defaultExecutor = new Executor()
    {
        final ArrayDeque<Runnable> tasks = new ArrayDeque<Runnable>();
        Runnable activeRunnable;

        @Override
        public void execute(final Runnable r)
        {
            tasks.offer(new Runnable()
            {

                @Override
                public void run()
                {
                    try
                    {
                        System.out.println(Thread.currentThread().getName());
                        r.run();
                    }
                    finally
                    {
                        scheduleNext();
                    }
                }
            });

            if(activeRunnable == null)
            {
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext()
        {
            if((activeRunnable = tasks.poll()) != null)
            {
                THREAD_POOL_EXECUTOR.execute(activeRunnable);
            }
        }
    };

    public void test()
    {
        List<FutureTask<String>> fList = new ArrayList<FutureTask<String>>();
        for(int i = 0; i < 10; i++)
        {
            final int j = i;

            fList.add(new FutureTask<String>(new Callable<String>()
            {

                @Override
                public String call() throws Exception
                {
                    Thread.sleep(3000);
                    return "I‘m callable " + j;
                }
            }){
                @Override
                protected void done()
                {
                    try
                    {
                        System.out.println(get() + " done");
                    }
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    catch (ExecutionException e)
                    {
                        e.printStackTrace();
                    }
                }
            }

            );
        }

        for(FutureTask<String> fTask : fList)
        {
            defaultExecutor.execute(fTask);
        }
    }

}

先看运行结果:
AsyncTask #1
I’m callable 0 done
AsyncTask #2
I’m callable 1 done
AsyncTask #3
I’m callable 2 done
AsyncTask #4
I’m callable 3 done
AsyncTask #5
I’m callable 4 done
AsyncTask #5
I’m callable 5 done
AsyncTask #5
I’m callable 6 done
AsyncTask #5
I’m callable 7 done
AsyncTask #5
I’m callable 8 done
AsyncTask #5
I’m callable 9 done
是不是一模一样~没错事实上我们正常调用AsyncTask的execute方法的时候,就是调用了这个defaultExecutor。它的作用就是维持了一个双向的任务队列,当AsyncTask的execute方法运行的时候,它就把client提交的任务塞到了这个队列里,假设这时候没有任务在运行。activeRunnable就为null,则scheduleNext方法直接调用,这个刚被提交的任务就会从队列中被取出交给线程池区运行,运行完毕后又会继续调用scheduleNext方法,有任务就会继续运行下一个任务。所以你看到的结果就是这样一个顺序运行。而且线程池仅仅使用了5个线程,充分利用了资源。补充一点,AsyncTask的源代码中,假设你想把全部任务改为并行运行,是能够传一个自己的Executor进来的。可是这种方法被hide了。看来是官方不建议大家这么做。

理解了上面两个样例的话。第三个样例写起来就so easy了,没错,理解轮子的最好试金石就是自己写个轮子。所以以下我们就是要简单地写一个自己的AsyncTask。和java直接run最大的差别就是安卓的非UI线程不能操作UI线程的实例,这个时候,把handler君请过来就好了嘛~ 还是先看代码。我们自己定义一个MyAsyncTask:

package com.amuro.thread;

import android.os.AsyncTask;
import android.os.Handler;
import android.os.Message;

import java.util.ArrayDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.LogRecord;

/**
 * Created by Echo on 2016/3/12.
 */
public abstract class MyAsyncTask<Params, Result>
{
    /*************线程池核心代码*******************/
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = CPU_COUNT + 1;
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
    private static final int KEEP_ALIVE = 1;

    private static final BlockingQueue<Runnable> workQueue =
            new LinkedBlockingQueue<Runnable>(10);
    private static final ThreadFactory threadFactory = new ThreadFactory()
    {
        private final AtomicInteger count = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r)
        {
            return new Thread(r, "MyAsyncTask #" + count.getAndIncrement());
        }
    };
    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR =
            new ThreadPoolExecutor(
                    CORE_POOL_SIZE,
                    MAXIMUM_POOL_SIZE,
                    KEEP_ALIVE,
                    TimeUnit.SECONDS,
                    workQueue,
                    threadFactory);

    private static volatile Executor defaultExecutor = new Executor()
    {
        final ArrayDeque<Runnable> tasks = new ArrayDeque<Runnable>();
        Runnable activeRunnable;

        @Override
        public void execute(final Runnable r)
        {
            tasks.offer(new Runnable()
            {

                @Override
                public void run()
                {
                    try
                    {
                        r.run();
                    }
                    finally
                    {
                        scheduleNext();
                    }
                }
            });

            if(activeRunnable == null)
            {
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext()
        {
            if((activeRunnable = tasks.poll()) != null)
            {
                THREAD_POOL_EXECUTOR.execute(activeRunnable);
            }
        }
    };

    /****************消息处理核心代码************************/
    private static final int MESSAGE_POST_RESULT = 0x01;

    private static class AsyncTaskResult<Data>
    {
        final MyAsyncTask mTask;
        final Data[] mData;

        AsyncTaskResult(MyAsyncTask task, Data... data)
        {
            mTask = task;
            mData = data;
        }
    }

    private static abstract class WorkerRunnable<Params, Result>
            implements Callable<Result> {
        Params[] mParams;
    }

    private static final Handler handler = new Handler()
    {
        @Override
        public void handleMessage(Message msg)
        {
            AsyncTaskResult result = (AsyncTaskResult) msg.obj;
            switch (msg.what)
            {
                case MESSAGE_POST_RESULT:
                    result.mTask.finish(result.mData[0]);
                    break;
            }
        }
    };
    private final WorkerRunnable<Params, Result> workerRunnable;
    private final FutureTask<Result> futureTask;

    public MyAsyncTask()
    {
        workerRunnable = new WorkerRunnable<Params, Result>()
        {
            @Override
            public Result call() throws Exception
            {
                return postResult(doInBackground(mParams));
            }
        };

        futureTask = new FutureTask<Result>(workerRunnable);
    }

    private Result postResult(Result result)
    {
        Message message = handler.obtainMessage(
                MESSAGE_POST_RESULT,
                new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }

    private void finish(Result result)
    {
        onPostExecute(result);
    }

    protected void onPreExecute(){}

    protected abstract Result doInBackground(Params... params);

    protected void onPostExecute(Result result){}

    public final MyAsyncTask<Params, Result> execute(Params... params)
    {
        return executeOnExecutor(defaultExecutor, params);
    }

    public final MyAsyncTask<Params, Result> executeOnExecutor(Executor executor, Params... params)
    {
        onPreExecute();

        workerRunnable.mParams = params;
        executor.execute(futureTask);
        return this;
    }
}

然后再看一下调用的代码:

package com.amuro.activity;

import android.app.Activity;
import android.os.AsyncTask;
import android.os.Bundle;
import android.util.Log;
import android.view.View;

import com.amuro.R;
import com.amuro.thread.MyAsyncTask;

public class MainActivity extends Activity
{

    @Override
    protected void onCreate(Bundle savedInstanceState)
    {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main_layout);

        findViewById(R.id.bt).setOnClickListener(new View.OnClickListener()
        {
            @Override
            public void onClick(View v)
            {
               testAsync();
            }

        });

        findViewById(R.id.bt1).setOnClickListener(new View.OnClickListener()
        {
            @Override
            public void onClick(View v)
            {
                testMyAsync();
            }
        });
    }

    private void testAsync()
    {
        for(int i = 0; i < 10; i++)
        {
            final int j = i;
            AsyncTask<String, Integer, String> aTask =
                    new AsyncTask<String, Integer, String>()
            {
                @Override
                protected void onProgressUpdate(Integer... values)
                {
                    super.onProgressUpdate(values);
                }

                @Override
                protected String doInBackground(String... params)
                {
                    Log.e("amuro", Thread.currentThread().getName());
                    try
                    {
                        Thread.sleep(1000);
                    }
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    return params[0] + "done";
                }

                @Override
                protected void onPostExecute(String s)
                {
                    Log.e("amuro", "result: " + s + " " + j);
                }
            };
            aTask.execute("DoubleX");
        }
    }

    private void testMyAsync()
    {
        for(int i = 0; i < 10; i++)
        {
            final int j = i;

            MyAsyncTask<String, String> myTask = new MyAsyncTask<String, String>()
            {
                @Override
                protected String doInBackground(String... params)
                {
                    Log.e("amuro", Thread.currentThread().getName());
                    try
                    {
                        Thread.sleep(1000);
                    }
                    catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                    return params[0] + "done";
                }

                @Override
                protected void onPostExecute(String s)
                {
                    Log.e("amuro", "result: " + s + " " + j);
                }
            };
            myTask.execute("outSideParam ");
        }
    }
}

再看一下运行结果:
03-13 13:15:55.065 20514-20732/com.amuro E/amuro: MyAsyncTask #1
03-13 13:15:56.070 20514-20514/com.amuro E/amuro: result: outSideParam done 0
03-13 13:15:56.070 20514-20747/com.amuro E/amuro: MyAsyncTask #2
03-13 13:15:57.075 20514-20514/com.amuro E/amuro: result: outSideParam done 1
03-13 13:15:57.075 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:15:58.075 20514-20514/com.amuro E/amuro: result: outSideParam done 2
03-13 13:15:58.075 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:15:59.075 20514-20514/com.amuro E/amuro: result: outSideParam done 3
03-13 13:15:59.075 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:16:00.080 20514-20514/com.amuro E/amuro: result: outSideParam done 4
03-13 13:16:00.080 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:16:01.080 20514-20514/com.amuro E/amuro: result: outSideParam done 5
03-13 13:16:01.080 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:16:02.080 20514-20514/com.amuro E/amuro: result: outSideParam done 6
03-13 13:16:02.080 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:16:03.085 20514-20514/com.amuro E/amuro: result: outSideParam done 7
03-13 13:16:03.085 20514-20758/com.amuro E/amuro: MyAsyncTask #3
03-13 13:16:04.085 20514-20514/com.amuro E/amuro: result: outSideParam done 8
03-13 13:16:04.090 20514-20732/com.amuro E/amuro: MyAsyncTask #1
03-13 13:16:05.095 20514-20514/com.amuro E/amuro: result: outSideParam done 9
暴露了我的測试机弱爆了。Orz。
为了简单起见这里就不处理onProgressUpdate了,有兴趣的同学能够在这个基础上自己去实现。

我在这里总结一下execute方法运行的整个流程。
1. 先回调了onPreExecute方法,这个是在UI线程里的。然后把外面传入的params赋值给了workerRunnable,事实上就是FutureTask须要的Callable对象。


2. 然后就把这个FutureTask丢给了我们的defaultExecutor去运行。这个流程和上面的样例二是一样一样的。


3. 运行成功后子线程完毕了结果的生成,这个时候就能够通过handler把结果丢给UI线程了。这里封装了一个AsyncTaskResult类来传递结果,原因非常easy。handler是静态对象。没法直接拿到当前MyAsyncTask的引用。而我们要把task和result对象同一时候丢给handler。所以要进行一下封装。
4. OK。handler拿到result之后就会把task拿出来并回调finish方法。
5. finish方法。这个时候已经在UI线程中了,所以能够回调终于的onPostExecute方法把结果丢给client去处理了。

不管多么复杂的技术或实现,仅仅要我们抓到其本质,耐心地把它涉及到的知识一点点的吃透,并多写代码多做測试。

终于你会发现,再复杂,只是也是小知识的层叠和扩展罢了。这和一个互联网公司须要深厚的技术积累,道理也是一样的。

就酱。谢谢欣赏~



教你写Http框架(二)——三个样例带你深入理解AsyncTask

原文地址:http://www.cnblogs.com/llguanli/p/7202135.html