设计模式 Concurrency 之 Half_Sync_Half_Async 半同步半异步模式

向右看齐 2022-06-10 03:14 200阅读 0赞

*  动机
 *  应用场景
 *  实现方案

# 1.动机 #

*  同步模式编程简单 但是IO的利用率低
 *  异步模式编程复杂 但是IO利用率高
 *  此模式综合了同步异步的优缺点 高层中使用同步IO模型,简化编程。低层使用异步IO模型 高效执行

# 2.应用场景 #

*  1.系统必须响应和处理外部异步发生的事件
 *  2、一个或多个任务必须在单独的控制线程中执行 其他任务可以在多线程执行  
    
    
     *  上层任务\[数据库查询,文件传输\]使用同步IO模式 简化编写并行程序的难度
     *  底层任务\[网络控制器的中断处理\]使用异步IO模型 提高了执行效率

# 3.实现方案 #

*  1.同步任务层  
    
    
     *  用户级的进程
     *  本层的任务完成上层的I/O操作，使用同步I/O模型，通过队列层的队列中传输数据。和异步层不同，同步层的任务使用活动对象执行，这些活动对象有自己运行栈和寄存器状态。当执行同步I/O的时候，他们会被阻塞/睡眠。
 *  2.队列层  
    
    
     *  这个层在同步任务层和异步任务层之间，提供了同步控制和缓存的功能。异步任务的I/O 事件被缓存到消息队列中，同步任务层在队列中提取这些事件（相反方向亦然）
 *  3.异步任务层  
    
    
     *  处理低层的事件，这些事件由多个外部的事件源产生（例如网卡，终端）。和异步任务不同，此层的实体是被动对象，没有自己的运行栈，要求不能被阻塞。
 *    
    ![这里写图片描述][SouthEast]

代码实现

AsyncTask:

package com.hqq.concurrency.half_syn_half_async;
    
    import java.util.concurrent.Callable;
    
    /** * AsyncTask * 封装了一些异步执行的操作和结果。这些操作一般是在后台线程中运行然后将结果通过回调方法传递给调用者 * Created by heqianqian on 2017/8/9. */
    public interface AsyncTask<T> extends Callable<T> {
    
        /** * 调用{ @link #call()}之前调用 比较大的任务不应该放在此处执行 很有可能会阻塞调用线程 * 适合一些小的诸如验证的任务 */
        void onPreCall();
    
        /** * 在{ @link #call()}调用之后执行. * @param result 程序执行结果 */
        void onPostCall(T result);
    
        /** * 执行中抛出异常的回调方法 * 当{ @link #call()}或者{ @link #onPreCall()}抛出异常时调用 * @param throwable 异常 */
        void onError(Throwable throwable);
    
    
        /** * 调用执行方法 * @return 返回结果 * @throws Exception 可能产生的异常 */
        @Override
        T call() throws Exception;
    }
    
    AsynchronousService 
    
    package com.hqq.concurrency.half_syn_half_async;
    
    import java.util.concurrent.*;
    
    /** * AsynchronousService * 这是一个异步请求层 当有新的请求到达时不会阻塞线程 * 它将请求传递给同步层 包括一个{ @link java.util.concurrent.BlockingQueue}队列和一个线程池 { @link ThreadPoolExecutor} * 线程池中的线程在后台同步执行完任务后通过回调方法把结果传递给调用者 * <p> * Created by heqianqian on 2017/8/9. */
    public class AsynchronousService { 
    
        /** * 充当本模式中的同步层 后台执行一些耗时的程序 */
        private ExecutorService service;
    
        /** * 使用{ @code wordQueue}连接同步层和异步层 * @param workQueue 工作队列 */
        public AsynchronousService(BlockingQueue<Runnable> workQueue) {
            service = new ThreadPoolExecutor(10, 10, 10, TimeUnit.SECONDS, workQueue);
        }
    
        public <T> void execute(final AsyncTask<T> task){
            try {
                //一些小任务的操作
                task.onPreCall();
            } catch (Exception e) {
                task.onError(e);
                return;
            }
            service.submit(new FutureTask<T>(task){
                @Override
                protected void done() {
                    super.done();
                    try {
                        task.onPostCall(get());
                    } catch (InterruptedException e) {
                    } catch (ExecutionException e) {
                        task.onError(e.getCause());
                    }
                }
            });
        }
    
    }

App

package com.hqq.concurrency.half_syn_half_async;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.LinkedBlockingQueue;
    
    /** * App * 半异步/半同步IO模式 * 适用场景: * 1.UNIX网络子系统 * 2.CORBA * 3.安卓异步框架 * Created by heqianqian on 2017/8/9. */
    public class App { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(App.class);
    
        public static void main(String... args) {
            AsynchronousService service  = new AsynchronousService(new LinkedBlockingQueue<>());
    
            service.execute(new ArithmeticSumTask(1000));
    
            service.execute(new ArithmeticSumTask(500));
            service.execute(new ArithmeticSumTask(2000));
            service.execute(new ArithmeticSumTask(1));;
        }
    
        static class ArithmeticSumTask implements AsyncTask<Long> {
    
            private long mills;
    
            public ArithmeticSumTask(long mills) {
                this.mills = mills;
            }
    
            @Override
            public void onPreCall() {
                if (mills < 0) {
                    throw new IllegalArgumentException("mills is less than 0");
                }
            }
    
            @Override
            public void onPostCall(Long result) {
                LOGGER.info(result.toString());
            }
    
            @Override
            public void onError(Throwable throwable) {
                throw new IllegalStateException("Should not occur");
            }
    
            @Override
            public Long call() throws Exception {
                return ap(mills);
            }
        }
    
        private static long ap(Long mills) {
            try {
                Thread.sleep(mills);
            } catch (InterruptedException e) {
                LOGGER.error("Exception caught.",e);
            }
            return mills * (mills + 1) / 2;
        }
    }

--------------------

[SouthEast]: /images/20220610/82168bc0094944b09cd04f34f87e20a6.png