异步框架asyn4j的原理

启动时调用init方法

[java] view plain copy

1. public void init(){
2. if (!run){
3. run = true;
4. //工作队列
5. workQueue = newPriorityBlockingQueue(maxCacheWork);
6. //是否存在工作队列满处理类
7. if (this.workQueueFullHandler != null) {
8. //自定义线程池
9. workExecutor = newAsynThreadPoolExecutor(work_thread_num, work_thread_num, 0L,TimeUnit.MILLISECONDS,
10. workQueue, new AsynThreadFactory(),new AsynWorkRejectedExecutionHandler(this.workQueueFullHandler),
11. executeWorkNum);
12. } else {
13. workExecutor = newAsynThreadPoolExecutor(work_thread_num, work_thread_num, 0L,TimeUnit.MILLISECONDS,
14. workQueue, new AsynThreadFactory(),executeWorkNum);
15. }
16. //回调函数工作队列
17. callBackQueue = newLinkedBlockingQueue();
18. //自定义回调函数线程池
19. callBackExecutor = newCallBackThreadPoolExecutor(callback_thread_num, callback_thread_num, 0L,
20. TimeUnit.MILLISECONDS, callBackQueue,new CallBackThreadFactory(),
21. new CallBackRejectedExecutionHandler(),callBackNum);
22. //service启动和关闭处理器
23. if (this.serviceHandler != null) {
24. this.serviceHandler.setServiceStat(0);
25. this.serviceHandler.setAsynService(this);
26. this.serviceHandler.process();
27. }
28. //启动工作队列满处理器
29. if (this.workQueueFullHandler != null) {
30. this.workQueueFullHandler.process();
31. }
32. //程序关闭后的处理
33. Runtime.getRuntime().addShutdownHook(newThread(){
34. public void run()
35. {
36. AsynServiceImpl.this.close(AsynServiceImpl.closeServiceWaitTime);
37. }
38. });
39. }
40. }

AsynServiceImpl主要有两个线程池和三个处理器组成

工作线程池AsynThreadPoolExecutor

[java] view plain copy

1. public classAsynThreadPoolExecutor
2. extends ThreadPoolExecutor
3. {
4. private AtomicLong executeWorkNum;

回调函数线程池CallbackThreadPoolExecutor

[java] view plain copy

1. public classCallBackThreadPoolExecutor
2. extends ThreadPoolExecutor
3. {
4. private AtomicLong callBackNum;

AsynThreadPoolExecutor的executeWorkNum，统计任务的执行数

AsynThreadPoolExecutor的callBackNum，统计回调函数的执行数

ServiceHandler处理器（启动或关闭）

WorkQueueFullHandler处理器（任务数超过maxCacheWork）

ErrorAsynWorkHandler处理器（任务执行异常）

执行时调用addWork方法

[java] view plain copy

1. public void addWork(Object tagerObject, Stringmethod, Object[] params, AsynCallBack asynCallBack, WorkWeight weight, booleancache)
2. {
3. if ((tagerObject == null) || (method ==null)) {
4. throw newIllegalArgumentException("target name is null or  target method name is null");
5. }
6. Object target = null;
7. //如果对象是String
8. if(tagerObject.getClass().isAssignableFrom(String.class)){
9. addWorkWithSpring(tagerObject.toString(),method, params, asynCallBack, weight);
10. return;
11. }
12. //如果对象是class
13. if ((tagerObject instanceof Class)) {
14. String classKey =((Class)tagerObject).getSimpleName();
15. //是否缓存，一个class缓存一个默认的对象，防止重复创建
16. if (cache)
17. {
18. target = targetCacheMap.get(classKey);
19. if (target == null)
20. {
21. target =newObject((Class)tagerObject);
22. targetCacheMap.put(classKey, target);
23. }
24. }
25. else
26. {
27. target = newObject((Class)tagerObject);
28. }
29. }
30. else
31. {
32. target = tagerObject;
33. }
34. if (target == null) {
35. throw newIllegalArgumentException("target object is null");
36. }
37. //封装成异步任务
38. AsynWork anycWork = newAsynWorkEntity(target, method, params, asynCallBack, weight);
39. addAsynWork(anycWork);
40. }

[java] view plain copy

1. public voidaddAsynWork(AsynWork asynWork){
2. if (!run) {
3. throw new Asyn4jException("asynservice is stop or no start!");
4. }
5. if (asynWork == null) {
6. throw newIllegalArgumentException("asynWork is null");
7. }
8. try<span style="font-family: Arial, Helvetica, sans-serif;">{</span>

[java] view plain copy

1. //通过semaphore来获取许可权限
2. if(this.semaphore.tryAcquire(addWorkWaitTime, TimeUnit.MILLISECONDS))
3. {
4. WorkProcessor workProcessor = newWorkProcessor(asynWork, this);
5. workExecutor.execute(workProcessor);
6. totalWork.incrementAndGet();
7. }
8. else{
9. log.warn("work queue is full,addwork to cache queue");
10. this.workQueueFullHandler.addAsynWork(asynWork);
11. }
12. }
13. catch (InterruptedException e)
14. {
15. log.error(e);
16. }
17. }

通过semaphore来控制最大的访问线程数，maxCacheWork就是semaphore的数量，也是工作队列的数量

这里的目的是控制工作队列的数量不能超过maxCacheWork（也可以通过用上限的queue来代替）

如果超过队列maxCacheWork，就用workQueueFullHandler去处理，处理方式同线程池的拒绝策略处理器（

newAsynWorkRejectedExecutionHandler(this.workQueueFullHandler)）不过线程池的拒绝策略没用到（前提是队列的有上限的队列），

工作任务WorkProcessor（内含AsynWorkEntity）

[java] view plain copy

1. public void run() {
2. Thread currentThread =Thread.currentThread();
3. if (this.asynWork.getThreadName() != null){
4. setName(currentThread,this.asynWork.getThreadName());
5. }
6. AsynCallBack result = null;
7. try{
8. result = this.asynWork.call();
9. if (result != null) {
10. ApplicationContext.getCallBackExecutor().execute(result);
11. }
12. }
13. catch (Throwable throwable){
14. if(this.applicationContext.getErrorAsynWorkHandler() != null) {
15. this.applicationContext.getErrorAsynWorkHandler().addErrorWork(this.asynWork,throwable);
16. }
17. }
18. finally{
19. this.applicationContext.getSemaphore().release();
20. }
21. }

首先修改线程名称，然后调用asynWork的call方法,如果有回调函数，就执行，如果有异常就执行ErrorAsynWorkHandler，最后Semaphore释放一个许可

AsynWorkEntity

[java] view plain copy

1. public AsynCallBack call()
2. throws Exception {
3. if (this.target == null) {
4. throw new RuntimeException("targetobject is null");
5. }
6. Class clazz = this.target.getClass();
7. String methodKey =MethodUtil.getClassMethodKey(clazz, this.params, this.method);
8. Method targetMethod =(Method)methodCacheMap.get(methodKey);
9. if (targetMethod == null){
10. targetMethod =MethodUtil.getTargetMethod(clazz, this.params, this.method);
11. if (targetMethod != null) {
12. methodCacheMap.put(methodKey,targetMethod);
13. }
14. }
15. if (targetMethod == null) {
16. throw newIllegalArgumentException("target method is null");
17. }
18. Object result =targetMethod.invoke(this.target, this.params);
19. if (this.anycResult != null) {
20. this.anycResult.setInokeResult(result);
21. }
22. return this.anycResult;
23. }

通过反射技术调用用具体方法，也用缓存技术，anycResult就是你定义的回调函数，没有就返回null

关闭时调用close方法

[java] view plain copy

1. publicvoid close(long waitTime){
2. if (run){
3. run = false;
4. try {
5. workExecutor.awaitTermination(waitTime,TimeUnit.MILLISECONDS);
6. callBackExecutor.awaitTermination(0L,TimeUnit.MILLISECONDS);
7. }
8. catch (InterruptedException e)
9. {
10. log.error(e);
11. }
12. //关闭工作线程和回调函数线程
13. workExecutor.shutdown();
14. callBackExecutor.shutdown();
15. //service关闭处理器
16. if (this.serviceHandler != null)
17. {
18. this.serviceHandler.setAsynWorkQueue(workQueue);
19. this.serviceHandler.setCallBackQueue(callBackQueue);
20. this.serviceHandler.setServiceStat(1);
21. this.serviceHandler.process();
22. }
23. }
24. }