Java Design Demo -简单的队列-异步多任务队列（java android）

简单的单线程队列 -- 工作的时候遇到劣质打印机。给打印机发消息，打印机就会打印，如果在打印机还在打印的时候，就

再发消息打印，就会出现消息丢失。所以需要给上一个任务一些处理的间隔时间.

单线程的消息队列示例

package demo1;
import java.util.LinkedList;
public class Main {
/**
* @param args
*/
private static Thread thread;
private static LinkedList<Runnable> list = new LinkedList<Runnable>();
static int test = 0;
public static void main(String[] args) {
// TODO Auto-generated method stub
final long time = System.currentTimeMillis();
for (int i = 0; i < 20; i++) {
tastEvent(new Runnable() {
public void run() {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out
.println("第"
+ (++test)
+ ("个任务耗时:" + (System
.currentTimeMillis() - time)));
}
});
}
}
public static void tastEvent(Runnable r) {
synchronized (list) {
list.add(r);
}
if (thread == null) {
thread = new Thread(run);
thread.start();
}
}
static Runnable run = new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
synchronized (list) {
while (!list.isEmpty()) {
// new Thread(list.poll()).start();
list.poll().run();
}
thread = null;
}
}
};
}

Java Design Demo -简单的队列-异步多任务队列（java android）

工作的时候遇到非常大的并发的情况，比如机器1秒只支持1000的并发，但是1秒接收了4000的并发。服务器就会崩掉。

最好将并发放到队列中，按1000的并发吞吐量来处理，这就是异步队列应用。

一个工程交给一个人做，需要花费3个月，交给2个人做，需要2个人做需要2个月，需要3个人做需要1个月半.....100.人.....1000人,几年也完不成。

带上以上道理看待以下的代码

观察以下代码（复制到android工程下运行）：最后耗时约1600毫秒而使用android的AsyncTask类来改写这段代码只需要耗时约200

final long timer=System.currentTimeMillis();
count=0;
final Handler h=new Handler();
for(int k=0;k<100;k++){
new Thread(){
@Override
public void run() {
// TODO Auto-generated method stub
try {
Thread.sleep(10);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
h.post(new Runnable() {
@Override
public void run() {
Toast.makeText(getApplicationContext()," 耗时"+ (System.currentTimeMillis() - timer), 1).show();
System.err.println("编号"+(count++)+"线程消耗了"+(System.currentTimeMillis()-timer));
}
});
}
}.start();

可见增加多线程不提高性能，反而因为系统在不同的线程之间切换降低效率。因此我们需要让线程有序执行任务

以下是异步多线程处理队列的demo

Java Design Demo -简单的队列-异步多任务队列（java android）

package demo2;
import demo2.Task.OnFinishListen;
public class Main {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
Task.setThreadMaxNum(3);
for (int i = 0; i < 15; i++) {
new Task() {
@Override
public Object obtainData(Task task, Object parameter)
throws Exception {
// TODO Auto-generated method stub
Thread.sleep(500);
return task.taskID;
}
}
.setOnFinishListen(new OnFinishListen() {
@Override
public void onFinish(Task task, Object data) {
// TODO Auto-generated method stub
System.err.println("任务编号"+task.taskID+"任务完成");
}
})
.setTaskID(i)
.start();
}
}
}

package demo2;
import java.util.HashMap;
import java.util.Map;
import java.util.Observable;
import java.util.Observer;
public abstract class Task<P,R> implements Runnable, Observer,TaskAction<P,R>{
//设置最大任务数
public static void setThreadMaxNum(int num) {
TaskQueue.ThreadMaxNum = num<1?1:num>100?100:num;
}
public static enum TaskPriority {
max, min;
}
/** 单例可以提高性能 */
protected final static Exception withoutException = new Exception(
"The state is without");
// 名字映射
private static HashMap<String, Task> nameTasks;
public static HashMap<String, Task> getNameTask() {
if (nameTasks == null) {
nameTasks = new HashMap<String, Task>();
}
return nameTasks;
}
public Task<P,R> setSingletonName(String singletonName) {
this.singletonName = singletonName;
return this;
}
public String getSingletonName() {
return singletonName;
}
public interface OnStartListen {
void onStart(Task t);
}
public interface OnProgressListen {
void onProgress(Task task, int progress, Object data);
}
public static interface OnFinishListen<P,R> {
void onFinish(Task<P,R> task, R data);
}
public interface OnSystemStartListen {
void onSystemStart(Task task);
}
public interface OnSystemFinishListen {
void OnSystemFinish(Task t, Object data);
}
/** 请求参数 */
protected P parameter;
/** 任务开始监听 */
protected OnStartListen onStartListen;
/** 任务进度监听 */
protected OnProgressListen onProgressListen;
/** 任务完成监听 */
protected OnFinishListen<P,R> onFinishListen;
/** 任务在队列中完成监听 */
protected OnSystemStartListen onSystemStartListen;
/** 任务在队列中开始监听 */
protected OnSystemFinishListen onSystemFinishListen;
/** 用于任务完成后发送消息 */
/** 结果 */
protected R result;
/** 任务编号标示 */
protected int taskID = -1;
/** 任务名字标示 */
/** 设置此任务名是否为单例，单例模式下，如果相同名字的任务未执行完，则无法添加新任务 */
protected String singletonName;
/** 保存一个对象 */
protected Object tag;
/** 获得当前自身线程的引用在threadRun方法 */
protected Thread thread;
/** 重连次数 */
protected int tryAgainCount = 1;
/** 重连间隔 */
protected int tryAgainTime = 1000;
/** 默认优先级低 */
protected TaskPriority priority = TaskPriority.min;
protected HashMap<String,Object> dataMap;
protected Task() {
}
// 任务状态
public static enum TaskStatus {
// 未处理出错完成执行中排除
untreated, wait,error, finsh, running, without;
}
/** 状态 */
TaskStatus status = TaskStatus.untreated;
public void setWithout() {
this.status = TaskStatus.without;
}
public void remove() {
this.status = TaskStatus.without;
}
public TaskPriority getPriority() {
return priority;
}
public void setPriority(TaskPriority priority) {
this.priority = priority;
}
/** 启动线程 */
public void start() {
if (this.priority == null)
this.priority = TaskPriority.min;
synchronized (TaskQueue.tasks_wait) {
if (getSingletonName() != null
&& Task.getNameTask().get(this.getSingletonName()) != null) {
this.setWithout();
} else {
Task.getNameTask().put(this.getSingletonName(), this);
}
switch (priority) {
case min:
TaskQueue.tasks_wait.remove(this);
TaskQueue.tasks_wait.add(this);
break;
case max:
TaskQueue.tasks_wait.remove(this);
TaskQueue.tasks_wait.addFirst(this);
break;
default:
break;
}
// 启动此服务
TaskQueue.serivesRun();
}
}
/** 启动线程 */
public void start(TaskPriority priority) {
this.priority = priority;
status=TaskStatus.wait;
start();
}
/** 启动线程 */
final void threadRun() {
thread = new Thread(this);
thread.start();
}
// 中断Execute方法
public void shutDownExecute(){};
public R cacheData(P parameter){
return result;};
// 禁止被重写
public final Object Execute() throws Exception {
// TODO Auto-generated method stub
if (onStartListen != null)
onStartListen.onStart(this);
// 队列中回调
if (onSystemStartListen != null)
onSystemStartListen.onSystemStart(this);
// 状态从未处理改变为处理中
status = TaskStatus.running;
// 获取最后一次是否错误
Exception exception = null;
// 是否有缓存数据如果没有
if ((result = cacheData(parameter)) == null) {
// 失败重联次数
for (int i = 0; i < tryAgainCount; i++) {
try {
// 如果状态改变为排除则跳出失败重联
if (status == TaskStatus.without) {
break;
}
exception = null;
result = obtainData(this, parameter);
System.out.println("result=" + result);
break;
} catch (Exception e) {
// TODO Auto-generated catch block
if ((exception = e) == withoutException) {
break;
}
e.printStackTrace();
try {
Thread.sleep(tryAgainTime);
} catch (Exception e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
}
}
// 如果最后一次仍然失败则抛出
if (exception != null) {
throw exception;
}
// 如果状态改变为处理完但不通知
if (status != TaskStatus.without) {
if (onFinishListen != null) {
//完成监听并将结果加入到主线程
onFinishListen.onFinish(this, result);
}
;
}
if (onSystemFinishListen != null) {
onSystemFinishListen.OnSystemFinish(this, result);
}
status = TaskStatus.finsh;
return result;
}
public abstract R obtainData(Task<P,R> task, P parameter)throws Exception;
@Override
public void update(Observable observable, Object data) {
// 移除观察
observable.deleteObserver(this);
// 中断停止关闭连接
this.shutDownExecute();
this.setWithout();
if (this.thread != null) {
this.thread.interrupt();
}
// 错误尝试次数为0
this.tryAgainCount = 0;
};
@Override
public void run() {
try {
Execute();
} catch (Exception e) {
e.printStackTrace();
status = TaskStatus.error;
// 如果状态改变为处理完但不通知
if (status != TaskStatus.without) {
if (onFinishListen != null) {
//将结果加入到主线程
onFinishListen.onFinish(this, result);
}
}
if (onSystemFinishListen != null) {
onSystemFinishListen.OnSystemFinish(this, e);
}
}
//递归避免新开线程唤醒等待中的任务
TaskQueue.getRunnable().notifyWaitingTask();
}
public Object getTag() {
return tag;
}
public Task setTag(Object tag) {
this.tag = tag;
return this;
}
public Thread getThread() {
return thread;
}
public TaskStatus getStatus() {
return status;
}
public Object getParameter() {
return parameter;
}
public Task setParameter(P parameter) {
this.parameter = parameter;
return this;
}
public OnStartListen getOnStartListen() {
return onStartListen;
}
public Task setOnStartListen(OnStartListen onStartListen) {
this.onStartListen = onStartListen;
return this;
}
public OnProgressListen getOnProgressListen() {
return onProgressListen;
}
public Task setOnProgressListen(OnProgressListen onProgressListen) {
this.onProgressListen = onProgressListen;
return this;
}
public OnFinishListen getOnFinishListen() {
return onFinishListen;
}
public Task setOnFinishListen(OnFinishListen onFinishListen) {
this.onFinishListen = onFinishListen;
return this;
}
public OnSystemStartListen getOnSystemStartListen() {
return onSystemStartListen;
}
public OnSystemFinishListen getOnSystemFinishListen() {
return onSystemFinishListen;
}
public void setOnSystemFinishListen(
OnSystemFinishListen onSystemFinishListen) {
this.onSystemFinishListen = onSystemFinishListen;
}
public int getTaskID() {
return taskID;
}
public Task setTaskID(int taskID) {
this.taskID = taskID;
return this;
}
public Object getResult() {
return result;
}
public int getTryAgainCount() {
return tryAgainCount;
}
public Task setTryAgainCount(int tryAgainCount) {
this.tryAgainCount = tryAgainCount;
return this;
}
public int getTryAgainTime() {
return tryAgainTime;
}
private Task setTryAgainTime(int tryAgainTime) {
this.tryAgainTime = tryAgainTime;
return this;
}
public Object put(String key,Object value) {
if(dataMap==null)
{
dataMap=new HashMap<String, Object>();
}
return dataMap.put(key, value);
}
public Object get(String key,Object value) {
if(dataMap==null)
{
dataMap=new HashMap<String, Object>();
}
return dataMap.get(key);
}
}

package demo2;
import java.util.AbstractCollection;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import demo2.Task.OnSystemFinishListen;
import demo2.Task.TaskStatus;
public class TaskQueue implements Runnable, OnSystemFinishListen {
static String debug = "TaskQueue";
@SuppressWarnings("unchecked")
// 在等待的任务队列
static LinkedList<Task> tasks_wait = new LinkedList<Task>();
public static class TaskQueueExpection extends Exception{
TaskQueueExpection(String detailMessage) {
super(detailMessage);
// TODO Auto-generated constructor stub
}
};
// 正在执行的任务
static ArrayList<Task> tasks_running = new ArrayList<Task>();
// 是否持续运行
public static boolean isRun=true;
// runnable保证线程安全
private static TaskQueue runnable = new TaskQueue();;
// 最大线程数
static int ThreadMaxNum = 1;
public static TaskQueue getRunnable() {
return runnable;
}
// 如果队列线程为空或者停止则重新开启
public static void serivesRun() {
// TODO Auto-generated method stub
boolean isCanSeriver=false;
synchronized (tasks_running) {
isCanSeriver=tasks_running.size() < ThreadMaxNum;
}
runnable.run();
}
//获取正在执行的任务数
public static int getRunningTaskCount() {
synchronized (TaskQueue.tasks_running) {
return TaskQueue.tasks_running.size();
}
}
//设置最大任务数
public static void setThreadMaxNum(int num) {
TaskQueue.ThreadMaxNum = num<1?1:num>100?100:num;
}
// 线程锁如果等待队列的任务数不为空，并且当前线程数字少于最大线程数
public static boolean taskRun() {
synchronized (tasks_wait) {
synchronized (tasks_running) {
return !tasks_wait.isEmpty()
&& tasks_running.size() < ThreadMaxNum;
}
}
}
//开启新线程
public void run() {
// 线程锁如果等待队列的任务数不为空，并且当前线程数字少于最大线程数
Task newTask;
while((newTask=getWaittingTask())!=null)
{
System.err.println("开启新线程处理一个新任务，ID："+newTask.getTaskID());
newTask.setOnSystemFinishListen(runnable);
newTask.threadRun();
newTask=null;
}
}
//递归避免新开线程唤醒等待中的任务但此方案会造成java.lang.*Error
void notifyWaitingTask()
{
Task newTask;
while((newTask=getWaittingTask())!=null)
{
System.err.println("唤醒旧线程处理一个新任务，ID："+newTask.getTaskID());
newTask.setOnSystemFinishListen(runnable);
newTask.run();
newTask=null;
}
}
private Task getWaittingTask()
{
Task t=null;
//测试
while (isRun && taskRun()) {
// 添加带执行中动态数组中
synchronized (tasks_wait) {
// 从等待任务的队列中获取并移除此列表的头（第一个元素）
t = tasks_wait.poll();
// 如果h为空则从队列重新取对象或者任务绑定的状态变化了
if (t == null || t.status == TaskStatus.without) {
System.out.println("任务取消编号" + t!=null?String.valueOf(t.getTaskID()):"空任务");
continue;
}
}
synchronized (tasks_running) {
tasks_running.add(t);
}
System.out.println( "正在执行任务数" + tasks_running.size() + "/上限"
+ ThreadMaxNum);
return t;
}
return t;
}
@Override
public void OnSystemFinish(Task t, Object data) {
// TODO Auto-generated method stub
synchronized (tasks_running) {
// 从处理中的动态数组中移除此任务
tasks_running.remove(t);
System.out.println( "执行队列中移除任务taskid=" + t.taskID);
// 通知执行后续未处理的任务
System.out.println("正在执行任务数" + tasks_running.size() + "/上限"
+ ThreadMaxNum);
// 移除此名字映射
if (t.getSingletonName() != null) {
Task.getNameTask().remove(t.getSingletonName());
}
}
}
}

Demo代码下载地址

秒客网

Java Design Demo -简单的队列-异步多任务队列（java android）

相关文章