高并发设计模式

高并发设计模式

 

 

懒汉式: 方法上加synchronized

public static synchronized Singleton getInstance() {  
         if (single == null) {    
             single = new Singleton();  
         }    
        return single;  
}  

懒汉式: 使用双检锁 + volatile

private volatile Singleton singleton = null;
    public static Singleton getInstance() {
        if (singleton == null) {
            synchronized (Singleton.class) {
                if (singleton == null) {
                    singleton = new Singleton();
                }
            }
        }
        return singleton;
    }

懒汉式: 使用静态内部类

public class Singleton {
    private static class LazyHolder {
       private static final Singleton INSTANCE = new Singleton();
    }
    private Singleton (){}
    public static final Singleton getInstance() {
       return LazyHolder.INSTANCE;
    }
}

===============》

饿汉式

public class Singleton1 {
    private Singleton1() {}
    private static final Singleton1 single = new Singleton1();
    public static Singleton1 getInstance() {
        return single;
    }
}

 

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

Future模式(该模式的核心思想是异步调用. 有点类似于异步的ajax请求.

通过FutureTask实现

public class FutureDemo1 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        FutureTask<String> future = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return new RealData().costTime();
            }
        });
        ExecutorService service = Executors.newCachedThreadPool();
        service.submit(future);

        System.out.println("RealData方法调用完毕");
        // 模拟主函数中其他耗时操作
        doOtherThing();
        // 获取RealData方法的结果
        System.out.println(future.get());
    }

    private static void doOtherThing() throws InterruptedException {
        Thread.sleep(2000L);
    }
}

class RealData {

    public String costTime() {
        try {
            // 模拟RealData耗时操作
            Thread.sleep(1000L);
            return "result";
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "exception";
    }

}

通过Future实现

public class FutureDemo2 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newCachedThreadPool();
        Future<String> future = service.submit(new RealData2());

        System.out.println("RealData2方法调用完毕");
        // 模拟主函数中其他耗时操作
        doOtherThing();
        // 获取RealData2方法的结果
        System.out.println(future.get());
    }

    private static void doOtherThing() throws InterruptedException {
        Thread.sleep(2000L);
    }
}

class RealData2 implements Callable<String>{

    public String costTime() {
        try {
            // 模拟RealData耗时操作
            Thread.sleep(1000L);
            return "result";
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "exception";
    }

    @Override
    public String call() throws Exception {
        return costTime();
    }
}

 

=====》Future本身还提供了一些额外的简单控制功能, 其API如下

// 取消任务
    boolean cancel(boolean mayInterruptIfRunning);
    // 是否已经取消
    boolean isCancelled();
    // 是否已经完成
    boolean isDone();
    // 取得返回对象
    V get() throws InterruptedException, ExecutionException;
    // 取得返回对象, 并可以设置超时时间
    V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;

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

生产消费者模式

生产者-消费者模式是一个经典的多线程设计模式. 它为多线程间的协作提供了良好的解决方案。

在生产者-消费者模式中,通常由两类线程,即若干个生产者线程和若干个消费者线程。

生产者线程负责提交用户请求,消费者线程则负责具体处理生产者提交的任务。

生产者和消费者之间则通过共享内存缓冲区进行通信

高并发设计模式

 

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

分而治之

Master-Worker模式

该模式核心思想是系统由两类进行协助工作: Master进程, Worker进程.

Master负责接收与分配任务, Worker负责处理任务. 当各个Worker处理完成后,

将结果返回给Master进行归纳与总结.

==》Master代码

public class MasterDemo {
    // 盛装任务的集合
    private ConcurrentLinkedQueue<TaskDemo> workQueue = new ConcurrentLinkedQueue<TaskDemo>();
    // 所有worker
    private HashMap<String, Thread> workers = new HashMap<>();
    // 每一个worker并行执行任务的结果
    private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();

    public MasterDemo(WorkerDemo worker, int workerCount) {
        // 每个worker对象都需要持有queue的引用, 用于领任务与提交结果
        worker.setResultMap(resultMap);
        worker.setWorkQueue(workQueue);
        for (int i = 0; i < workerCount; i++) {
            workers.put("子节点: " + i, new Thread(worker));
        }
    }

    // 提交任务
    public void submit(TaskDemo task) {
        workQueue.add(task);
    }

    // 启动所有的子任务
    public void execute(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            entry.getValue().start();
        }
    }

    // 判断所有的任务是否执行结束
    public boolean isComplete() {
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            if (entry.getValue().getState() != Thread.State.TERMINATED) {
                return false;
            }
        }

        return true;
    }

    // 获取最终汇总的结果
    public int getResult() {
        int result = 0;
        for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
            result += Integer.parseInt(entry.getValue().toString());
        }

        return result;
    }

}

 

=======》Worker代码

public class WorkerDemo implements Runnable{

    private ConcurrentLinkedQueue<TaskDemo> workQueue;
    private ConcurrentHashMap<String, Object> resultMap;

    @Override
    public void run() {
        while (true) {
            TaskDemo input = this.workQueue.poll();
            // 所有任务已经执行完毕
            if (input == null) {
                break;
            }
            // 模拟对task进行处理, 返回结果
            int result = input.getPrice();
            this.resultMap.put(input.getId() + "", result);
            System.out.println("任务执行完毕, 当前线程: " + Thread.currentThread().getName());
        }
    }

    public ConcurrentLinkedQueue<TaskDemo> getWorkQueue() {
        return workQueue;
    }

    public void setWorkQueue(ConcurrentLinkedQueue<TaskDemo> workQueue) {
        this.workQueue = workQueue;
    }

    public ConcurrentHashMap<String, Object> getResultMap() {
        return resultMap;
    }

    public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
        this.resultMap = resultMap;
    }
}

主函数测试

MasterDemo master = new MasterDemo(new WorkerDemo(), 10);
        for (int i = 0; i < 100; i++) {
            TaskDemo task = new TaskDemo();
            task.setId(i);
            task.setName("任务" + i);
            task.setPrice(new Random().nextInt(10000));
            master.submit(task);
        }

        master.execute();

        while (true) {
            if (master.isComplete()) {
                System.out.println("执行的结果为: " + master.getResult());
                break;
            }
        }

==========。》

ForkJoin线程池

其核心思想也是将任务分割为子任务,

有可能子任务还是很大, 还需要进一步拆解, 最终得到足够小的任务.

将分割出来的子任务放入双端队列中, 然后几个启动线程从双端队列中获取任务执行.

子任务执行的结果放到一个队列里, 另起线程从队列中获取数据, 合并结果.

public class CountTask extends RecursiveTask<Long>{
    // 任务分解的阈值
    private static final int THRESHOLD = 10000;
    private long start;
    private long end;


    public CountTask(long start, long end) {
        this.start = start;
        this.end = end;
    }

    public Long compute() {
        long sum = 0;
        boolean canCompute = (end - start) < THRESHOLD;
        if (canCompute) {
            for (long i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 分成100个小任务
            long step = (start + end) / 100;
            ArrayList<CountTask> subTasks = new ArrayList<CountTask>();
            long pos = start;
            for (int i = 0; i < 100; i++) {
                long lastOne = pos + step;
                if (lastOne > end) {
                    lastOne = end;
                }
                CountTask subTask = new CountTask(pos, lastOne);
                pos += step + 1;
                // 将子任务推向线程池
                subTasks.add(subTask);
                subTask.fork();
            }

            for (CountTask task : subTasks) {
                // 对结果进行join
                sum += task.join();
            }
        }
        return sum;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        // 累加求和 0 -> 20000000L
        CountTask task = new CountTask(0, 20000000L);
        ForkJoinTask<Long> result = pool.submit(task);
        System.out.println("sum result : " + result.get());
    }
}

上一篇:chrome浏览器的console检测xpath定位是否正确


下一篇:使用自己的CSS框架(转)