频道栏目
首页 > 程序开发 > 移动开发 > Android > 正文
android网络业务的封装与调度
2011-12-17 15:28:48      个评论      
收藏   我要投稿

 

手机客户端程序由于网络宽带的约束,尤其在GPRS网络环境下,大数据量的网络交互很大程度上降低应用的响应,影响用户体验。比如,如果做一个手机网盘客户端,在后台上传文件时(大数据量的交互),获取文件列表(命令类的交互)这个过程就显得太别慢。而我们的要求是希望这些命令类操作能尽快得到响应。

 

通常,在手机客户端,我们设计一个网络操作的管理器,来统一管理这些需要联网的操作。

具体做法是把网络操作封装成一个Command(或者说是Task),管理器实现特定的调度规则来调度运行这些Task。

这样做的好处至少有三:

一. 用Command封装了网络操作,使得这些操作与上传的业务分离,解除了强耦合。

二. 可以根据网络情况来确定来采用不同的调度规则,提高用户体验。

三. 重用,这些Task和TaskManager的代码在别的手机应用上基本上能照搬过去。

四. 扩展,当应用需要扩展新的业务时,只有扩展一个新的Command(或者说是Task),接受调度即可,易于扩展。

 

例子:

还是以上文提到的微盘为例,可以概括我们对管理器的设计要求有:

在Wifi网络环境下:

一:各种网络操作可以并行运行。

在GPRS网络环境下:

二:支持优先级抢占调度,命令类操作的优先级比数据传输类的优先级高,当命令类的Task(获取文件列表)提交后,打断数据传输的Task(如上传,下载),等命令类的任务运行完毕,再接着运行数据类任务(断点上传,下载)。

二:同一个优先级的任务可以并行运行,如多个命令一起在网络上传输。

 

实现思路:

TaskManager

1. TaskManager开辟一个后台线程进行调度工作。

2. 由于要支持多个优先级的抢占调度,我们需要两个队列来维护运行中的Task和等待中的Task。

3. 由于Task的调度是基于优先级的,我们可以使用优先级队列,运行队列采用PriorityQueue,等待队列使用PriorityBlockingQueue,当没有网络业务需要运行时,调度线程阻塞挂起,避免空转。

4. TaskManager设计为单一实例(单一模式)。

5. 每个Task被调度运行时,该Task被从等待队列移动运行队列,当Task执行完毕时,从运行队列删除,唤醒调度线程进行新的调度。

下面是简单的设计代码

www.2cto.com

 

public final class TaskEngine implements Runnable{

 

private PriorityQueue<Task> runningQueue;//运行的task队列

 

private PriorityBlockingQueue<Task> readyQueue;//就绪的task队列,准备接受调度的task列表

 

private final AtomicLong taskIdProducer = new AtomicLong(1);//Task Id生成器

 

private Object sheduleLock = new Object();//同步锁

 

private static TaskEngine instance;

 

public long addTask(BusinessObject bo){

  Task task = new Task(bo);

  long newTaskId = taskIdProducer.incrementAndGet();

  task.setTaskId(newTaskId);

  if(this.isWifiNetWork()){ //WIFI网络

  synchronized(sheduleLock){

  runningQueue.add(task);

  }

  new Thread(task).start();

  }else{ //GPRS网络

  if(readyQueue.offer(task)){ //task入就绪队列

  final ReentrantLock lock = this.lock;

  lock.lock();

  try{

  needSchedule.signal(); //唤醒调度线程重新调度

  }finally{

  lock.unlock();}

  }

}

return newTaskId;

}

 

public final void run(){//task调度逻辑

....

....

}

 

//挂起调度线程 当不需要调度时

private void waitUntilNeedSchedule() throws InterruptedException

{

.....

}

}

 

Task:

 

1. 对要执行的网络操作的封装。

 

2. Task执行完毕时,发Task结束信号,唤醒调度线程进行新的调度

 

3. Task要实现Comparable接口,才能让TaskManager的两个队列自动其包含的Task排序。

 

下面是简单的设计代码:

 

public class Task implements Runnable,Comparable<Task>{

 

private long taskId;

 

private BusinessObject bo;//封装网络操作的业务对象的抽象父类,

//它封装了具体的Command,保证了业务扩展中Task的接口不变

 

@Override

public void run() {

  this.onTaskStart();

  this.bo.execute();

  this.onTaskEnd();

}

 

private voidonTaskStart()

{...}

 

public int getPriority()

{...}

 

public void setPriority(intpriority)

{...}

 

@Override

public int compareTo(Task object1) {

return this.getPriority()>object1.getPriority()?-1:1;

}

} 

 

 

 

小注意事项:

 

Android对PriorityQueue的实现和Jdk中的实现有点不一样。

 

(PriorityQueue.java在android中的代码路径是usr\dalvik\libcore\luni\src\main\java\java\util)

 

PriorityQueue.remove(Object o) ;//从PriorityQueue中删除一个元素。

 

对于完成这个删除操作android和jdk都是分两个过程实现,一,找出待删除元素的索引index,二,删除index所在元素。

 

在JDK中,是通过调用元素的equals方法来找到待删除元素的索引,

 

private int indexOf(Object o) {

if (o != null) {

for (int i = 0; i < size; i++)

if (o.equals(queue[i]))

return i;

}

return -1;

}

 

在android中,是间接调用元素的compareTo方法判断结果是否为0来找到待删除元素的索引,

 

int targetIndex;

for (targetIndex = 0; targetIndex < size; targetIndex++) {

if (0 == this.compare((E) o, elements[targetIndex])) {

break;

}

}

 

private int compare(E o1, E o2) {

if (null != comparator) {

return comparator.compare(o1, o2);

}

return ((Comparable<? super E>) o1).compareTo(o2);

}

 

所以为了Task能在执行完毕时从PriorityQueue找到这个Task并删除之,需要在compareTo方法里在优先级相等时

 

返回0。

 

@Override

public int compareTo(Task object1) {

if(this.getPriority()==object1.getPriority())

return 0;

return this.getPriority()>object1.getPriority()?-1:1;

}

 

当是这样运行PriorityQueue.remove(Object o) 逻辑上只能删除PriorityQueue里第一个优先级与被删除的元素

 

优先级相等的元素(可能是待删除的元素也可能不是),有误删的可能,需要做如下修改:

 

@Override

public int compareTo(Task object1) {

if(this.getPriority()==object1.getPriority() && this.equals(object1))

return 0;

return this.getPriority()>object1.getPriority()?-1:1;

}

 

这样才能正确执行remove(Object o),删除指定的对象o。

 

 

 

个人觉得android这样设计使得remove(Object o)复杂化了,不然JDK中那么简洁。语义上也不是那么好懂了,

 

因为按通常理解,equals是判断两个对象是否相等,compareTo可能是对象的某个属性的比较(类别数据库中的order by),

 

而现在执行PriorityQueue.remove(Object o),这个对象o明明在容器PriorityQueue中,却删除不了,除非去翻看android中PriorityQueue的实现代码,然后重写compareTo这个方法。这样的API使用时比较容易出错,应该不符号良好的API设计规范

 

吧。

 

完整的代码实现如下:

 

View Code

  1 * @类名:TaskEngine

  2  * @创建:baiyingjun (devXiaobai@gmail.com)

  3  * @创建日期:2011-7-7

  4  * @说明:task调度引擎

  5  ***************************************************/

  6 public final class TaskEngine implements Runnable{

  7

  8     private static final String TAG=Log.makeTag(TaskEngine.class);

  9    

 10     private PriorityQueue<Task> runningQueue;//运行的task队列

 11    

 12     private PriorityBlockingQueue<Task> readyQueue;//就绪的task队列,准备接受调度的task列表

 13    

 14     private final AtomicLong taskIdProducer = new AtomicLong(1);

 15    

 16     private Object sheduleLock = new Object();//调度锁

 17    

 18     private static TaskEngine instance;

 19    

 20     private final ReentrantLock lock = new ReentrantLock(true);

 21    

 22     private final Condition needSchedule = lock.newCondition();

 23    

 24     private Task currentTask;//准备接受调度的task

 25    

 26     private Context mAppContext;

 27    

 28     /**

 29      * add BusinessObject to taskEngine

 30 */

 31     public long addTask(BusinessObject bo) throws NetworkNotConnectException{

 32         Task task = new Task(bo);

 33         long newTaskId = taskIdProducer.incrementAndGet();

 34         task.setTaskId(newTaskId);

 35         if(Log.DBG){

 36             Log.d(TAG, "Add task with task id "+newTaskId+", priority "+task.getPriority());

 37         }

 38        

 39         if(this.isWifiNetWork()){ //WIFI网络

 40             synchronized(sheduleLock){

 41                 runningQueue.add(task);

 42             }

 43             new Thread(task).start();

 44         }else{ //GPRS网络

 45             if(readyQueue.offer(task)){ //task入就绪队列

 46                 if(Log.DBG)

 47                     Log.d(TAG, "add task " +task.bo.methodName+" "+task.taskId+" to ready queue");

 48                 final ReentrantLock lock = this.lock;

 49                 lock.lock();

 50                 try{

 51                     needSchedule.signal(); //唤醒调度线程重新调度

 52                 }finally{

 53                     lock.unlock();

 54                 }

 55             }

 56             //schedule();

 57         }

 58         return newTaskId;

 59     }

 60    

 61     private TaskEngine(Context context){

 62         mAppContext = context;

 63         runningQueue = new PriorityQueue<Task>();

 64         readyQueue = new PriorityBlockingQueue<Task>();

 65         new Thread(this).start();

 66         Log.i(TAG, "shedule thread working");

 67     }

 68    

 69     public synchronized static TaskEngine getInstance(Context context){

 70         Context appContext = context.getApplicationContext();

 71         if(instance==null || instance.mAppContext!=appContext){

 72             instance=new TaskEngine(appContext);

 73         }

 74         return instance;

 75     }

 76    

 77     protected boolean isWifiNetWork() throws NetworkNotConnectException{

 78         return NetworkManager.isWIFINetWork(mAppContext);

 79     }

 80    

 81     /**

 82      * task调度逻辑

 83 */

 84     public final void run(){

 85         Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);

 86         while(true){

 87             try {

 88                 if(this.isWifiNetWork()){

 89                     Task task = this.readyQueue.take();

 90                     if(task !=null){

 91                         synchronized(sheduleLock){

 92                             runningQueue.add(task);

 93                         }

 94                         new Thread(task).start();

 95                     }

 96                 }

 97                 else{//非wifi网络

 98 //空就绪队列,空运行队列,等待直到有任务到来

 99                     if(this.readyQueue.size()==0 && runningQueue.size()==0){

100                         currentTask=readyQueue.take();

101                         synchronized(sheduleLock){

102                             runningQueue.add(currentTask);

103                         }

104                         new Thread(currentTask).start();

105                     }

106                     //抢占式调度(就绪队列非空,运行队列优先级比就绪队列优先级低)

107                     else if(readyQueue.size()>0 &&

108                             this.readyQueue.element().getPriority()>=this.getMaxPriority()){

109                         currentTask = readyQueue.take();

110                         if(currentTask.getPriority()>this.getMaxPriority()){//暂停低优先级的任务运行

111                             synchronized(sheduleLock){

112                                 for(int i=0;i<runningQueue.size();i++){

113                                     Task toStopTask =runningQueue.remove();

114                                     //因为任务调度,将低优先级的任务暂时给冻结起来

115                                     toStopTask.setState(Task.STATE_FROST);

116                                     readyQueue.add(toStopTask);

117                                 }

118                             }

119                         }

120                         //运行被调度的任务

121                         runningQueue.add(currentTask);

122                         new Thread(currentTask).start();

123                     }else {//等高优先级的任务运行完毕

124                         waitUntilNeedSchedule();

125                     }

126                 }

127            

128             }catch (InterruptedException e) {

129                 Log.e(TAG, "Schedule error "+e.getMessage());

130             } catch (NetworkNotConnectException e) {

131                 // TODO Auto-generated catch block

132                 e.printStackTrace();

133             }

134         }

135     }

136

137     /*

138      * 等待,直到就绪队列里的最高优先级比当前运行优先级高,或者就绪队列为空时等待运行队列运行完毕

139 */

140     private void waitUntilNeedSchedule() throws InterruptedException{

141          final ReentrantLock lock = this.lock;

142          lock.lockInterruptibly();

143             try {

144                 try{

145                     while ((readyQueue.size()>0

146                             && readyQueue.element().getPriority()<getMaxPriority())//等高优先级的任务运行完毕

147                             || (readyQueue.size()==0 && runningQueue.size()>0) ){//或者等运行队列运行完毕

148                         if(Log.DBG)

149                             Log.d(TAG, "waiting sheduling........");

150                         needSchedule.await();

151                     }

152                 }

153                 catch (InterruptedException ie) {

154                     needSchedule.signal(); // propagate to non-interrupted thread

155                     throw ie;

156                 }

157             } finally {

158                 lock.unlock();

159             }

160     }

161    

162     /**

163      * Hand the specified task ,such as pause,delete and so on

164 */

165     public boolean handTask(long taskId,int handType) {

166         Log.i(TAG, "set task`s state with taskId "+taskId);

167         synchronized(this.sheduleLock){

168             //如果在运行队列里,取消该任务

169             Iterator<Task> runningItor= this.runningQueue.iterator();

170             while(runningItor.hasNext()){

171                 Task task = runningItor.next();

172                 boolean b = task.equals(this);

173                 if(task.getTaskId()==taskId){

174                     runningQueue.remove(task);

175                     task.setState(handType);

176                     Log.i(TAG, "set runningQueue taskId = "+taskId + " state " + handType);

177                     return true;

178                 }

179             }

180             //如果在就绪队列里,删除

181             Iterator<Task> readyItor= this.readyQueue.iterator();

182             while(readyItor.hasNext()){

183                 Task task = readyItor.next();

184                 if(task.getTaskId()==taskId){

185 //                    readyQueue.remove(task);

186                     task.setState(handType);

187                     Log.i(TAG, "set readyQueue taskId = "+taskId + " state " + handType);

188                     return true;

189                 }

190             }

191             return false;

192         }

193     }

194     /***

195      * 获取运行队列任务的最高优先级

196 */

197     private int getMaxPriority(){

198         if(this.runningQueue==null || this.runningQueue.size()==0)

199             return -1;

200         else{

201             return this.runningQueue.element().getPriority();

202         }

203     }

204        

205     /***************************************************

206      * @类名:Task

207      * @创建:baiyingjun (devXiaobai@gmail.com)

208      * @创建日期:2011-7-7

209      * @说明:业务对象的包装成可运行实体

210      ***************************************************/

211     public class Task implements Runnable,Comparable<Task>{

212

213         //运行

214         public static final int STATE_INIT = -1;

215         //运行

216         public static final int STATE_RUN = 0;

217         //停止

218         public static final int STATE_STOP = 1;

219         //暂停

220         public static final int STATE_PAUSE = 2;

221         //取消

222         public static final int STATE_CANCLE = 3;

223         //冻结,如果在GPRS下,因为线程调度的时候低优先级的被放readyqueue里的时候,要把这个任务暂时给“冻结”起来

224         public static final int STATE_FROST = 4;

225        

226         private long taskId;

227        

228         private BusinessObject bo;

229        

230         public Task(){

231            

232         }

233        

234         public Task(BusinessObject bo){

235             this.bo=bo;

236         }

237        

238         @Override

239         public void run() {

240             this.onTaskStart();

241             this.bo.execute();

242             this.onTaskEnd();

243         }

244

245         private void onTaskStart(){

246             this.bo.setmState(STATE_RUN);

247         }

248        

249         public long getTaskId() {

250             return taskId;

251         }

252

253         public void setTaskId(long taskId) {

254             this.taskId = taskId;

255         }

256

257         public int getPriority() {

258             return this.bo.getPriority();

259         }

260

261         public void setPriority(int priority) {

262             this.bo.setPriority(priority);

263         }

264

265         /*

266          * compare task priority

267 */

268         @Override

269         public int compareTo(Task object1) {

270             if(this.getPriority()==object1.getPriority()&& this.equals(object1))

271                 return 0;

272             return this.getPriority()>object1.getPriority()?-1:1;

273         }

274        

275         public void setState(int state){//设置当前运行的task的state

276             this.bo.setmState(state);

277             Log.d(TAG, "Set task "+this.bo.methodName+" "+this.taskId + " state " + state);

278         }

279        

280         private void onTaskEnd(){//运行完毕后从taskengine运行队列里删除

281             if(Log.DBG){

282                 Log.d(TAG, "task "+this.bo.methodName+" "+taskId+" End");

283             }

284             if(this.bo.getmState() == STATE_FROST)//因为调度停止了该业务

285                 return;

286            

287             final ReentrantLock lock = TaskEngine.this.lock;

288             lock.lock();

289             try{

290                 boolean removed = runningQueue.remove(this); //remove from running queue

291                 assert removed;

292                 if(Log.DBG)

293                     Log.d(TAG, this.bo.methodName+" "+this.taskId+" remove from runningQueue");

294                 needSchedule.signal(); //唤醒调度线程重新调度

295             }finally{

296                 lock.unlock();

297             }

298         }

299

300         @Override

301         public boolean equals(Object o) {

302             // TODO Auto-generated method stub

303             if(this==o){

304                 return true;

305             }

306             if(o instanceof Task){

307                 return taskId==((Task)o).taskId;

308             }

309             return false;

310         }

311     }

312    

313 }完

 

拾漏补遗:

1.补充最初的设计类图(可能与代码不太一致,但能说明问题)

\

2. BusinessObject的实现代码:

 

 1 /***************************************************

 2  * @类名:BusinessObject

 3  * @创建:baiyingjun (devXiaobai@gmail.com)

 4  * @创建日期:2011-7-6

 5  * @说明:抽象的业务,扩展的网络业务要继承这个类并实现抽象方法execute()

 6  ***************************************************/

 7 public abstract class BusinessObject {

 8    

 9     public static final int LOWEST_PRIORITY=1;//最低优先级

10    

11     public static final int LOW_PRIORITY=2;//低优先级

12    

13     public static final int NORMAL_PRIORITY=3;//正常优先级

14    

15     public static final int HIGH_PRIORITY=4;//高优先级

16    

17     public static final int HIGHEST_PRIORITY=5;//最高优先级

18    

19     protected BusinessListener listnener;//运行业务的回调

20    

21     protected Context mContext;

22    

23     private long taskId;

24    

25     protected Map<String,Object> params;//运行业务需要的参数

26    

27     private int priority;

28    

29     public int getPriority() {

30         return priority;

31     }

32

33     public void setPriority(int priority) {

34         this.priority = priority;

35     }

36

37     //设置回调

38     public void addBusinessListener(BusinessListener listnener){

39         this.listnener=listnener;

40     }

41    

42     public long doBusiness() throws NetworkNotConnectException{

43         taskId= TaskEngine.getInstance(mContext).addTask(this);

44         return taskId;

45     }

46    

47     public abstract void execute();

48 }

作者 白 英军

点击复制链接 与好友分享!回本站首页
相关TAG标签 业务 网络
上一篇:分享10种可提升Android应用运行效果的技巧
下一篇:AIDL学习笔记2之从Service获取地理位置
相关文章
图文推荐
点击排行

关于我们 | 联系我们 | 广告服务 | 投资合作 | 版权申明 | 在线帮助 | 网站地图 | 作品发布 | Vip技术培训 | 举报中心

版权所有: 红黑联盟--致力于做实用的IT技术学习网站