并发编程学习

JAVA同步类容器线程是否安全以及介绍：

1，ArrayList,LinkedList线程不安全
Vector 对应 CopyOnWriteArrayList 线程安全
2，hashmap线程不安全
hashtable 对应 ConcurrentHashMap 线程安全
3，Queue下的阻塞队列和非阻塞队列：

ConcurrentLinkedQueue高性能，高并发，线程安全的非阻塞队列。

下面是阻塞队列：


多线程设计模式

1，future模式
代码demo
package com.wjb.demo.futuredemo;

/**
 * Created by wjb on 2018/1/10.
 */
public interface Data {
    String getRequest();
}
=================================================
package com.wjb.demo.futuredemo;

/**
 * Created by wjb on 2018/1/10.
 */
public class FutureClient {
    public Data request(String queryString) {
        //返回futureData包装类，此时暂无数据
        final FutureData futureData = new FutureData();
        new Thread(new Runnable() {
            @Override
            public void run() {
                RealData realData = new RealData(queryString);
                futureData.setRealData(realData);
            }
        }).start();
        return futureData;
    }
}
===========================================
package com.wjb.demo.futuredemo;

/**
 * Created by wjb on 2018/1/10.
 */
public class FutureData implements Data {
    private RealData realData;
    private boolean isReady = false;

    public synchronized void setRealData(RealData realData) {
        //如果已经装载完毕直接返回
        if (isReady) {
            return;
        }
        this.realData=realData;
        isReady=true;
        notify();
    }


    @Override
    public synchronized String getRequest() {
        //如果没有装载好，一直阻塞
        while (!isReady){
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //如果装载好了返回真实数据,这里调用的是真实数据realData的getRequest()方法。
        return this.realData.getRequest();
    }
}
=================================================
package com.wjb.demo.futuredemo;

/**
 * Created by wjb on 2018/1/10.
 * 真实数据类
 */
public class RealData implements Data {
    private String result;
    public RealData(String queryString){
        System.out.println("根据"+queryString+"查询操作，一系列的操作省略。。。" );
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("操作完成");
        result="查询结果";
    }

    @Override
    public String getRequest() {
        return result;
    }
}
================================================
package com.wjb.demo.futuredemo;

/**
 * Created by wjb on 2018/1/10.
 * 多线程Future模式 示例
 */
public class Main {
    public static void main(String[] args) {
        FutureClient client = new FutureClient();
        //这里返回的暂时是空数据
        Data data = client.request("查询请求");
        System.out.println("请求发送成功");
        System.out.println("数据正在处理中，此时可以做其它操作");
        String result = data.getRequest();
        System.out.println(result);
    }
}
2，Master-Worker模式：
示例demo:
package com.wjb.demo.master_worker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by wjb on 2018/1/10.
 */
public class Master {
    //需要一个装任务的容器，这里使用非阻塞队列
    private ConcurrentLinkedQueue<Task> queue = new ConcurrentLinkedQueue();
    //需要一个装worker的容器
    private HashMap<String,Thread> map = new HashMap<String,Thread>();
    //需要一个容器放所有worker处理完的数据
    private ConcurrentHashMap<String,Task> resultMap = new ConcurrentHashMap<String,Task>();

    public Master(Worker worker,int count){
        worker.setMap(resultMap);
        worker.setQueue(queue);
        for(int i =0;i<count;i++){
            this.map.put(Integer.toString(i),new Thread(worker));
        }
    }
    public void submit(Task task){
        this.queue.add(task);
    }

    public void execute(){
        for (Map.Entry<String,Thread> m:map.entrySet()){
            m.getValue().start();
        }
    }

    public boolean isComplete(){
        for (Map.Entry<String,Thread> m:map.entrySet()){
            if (m.getValue().getState() != Thread.State.TERMINATED){
                return false;
            }
        }
        return true;
    }

}
=================================================
package com.wjb.demo.master_worker;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by wjb on 2018/1/10.
 */
public class Worker implements Runnable{
    private ConcurrentLinkedQueue<Task> queue;
    private ConcurrentHashMap<String,Task> map;

    public void setQueue(ConcurrentLinkedQueue queue){
        this.queue=queue;
    }
    public void setMap(ConcurrentHashMap<String,Task> map){
        this.map=map;
    }

    @Override
    public void run() {
        boolean flag = true;
        while (flag){
            Task task = this.queue.poll();
            if(task == null){
                break;
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            this.map.put(Integer.toString(task.getId()),task);
        }
    }
}
================================================
package com.wjb.demo.master_worker;

/**
 * Created by wjb on 2018/1/10.
 */
public class Task {
    private int id;
    private int price;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public int getPrice() {
        return price;
    }

    public void setPrice(int price) {
        this.price = price;
    }
}
=================================================
package com.wjb.demo.master_worker;

import java.util.Random;

/**
 * Created by wjb on 2018/1/10.
 * 多线程master-workersa模式
 */
public class Main {
    public static void main(String[] args) {
        Master master = new Master(new Worker(), 10);

        Random random = new Random();
        for(int i = 0;i<100;i++){
            Task task = new Task();
            task.setId(i);
            task.setPrice(random.nextInt());
            master.submit(task);
        }
        master.execute();
        long start = System.currentTimeMillis();
        while (true){
            if (master.isComplete()){
                long end = System.currentTimeMillis();
                System.out.println("耗时："+(end-start));
                break;
            }
        }
    }
}
3,生产者消费者模式：

示例demo

package com.wjb.demo.provider_consumer;

import java.util.concurrent.*;

/**
 * Created by wjb on 2018/1/11.
 */
public class Main {
    public static void main(String[] args) {
        //无界的阻塞队列，个数无限制
        BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>();

        //缓存线程池，可以创建无限大数量的线程，没有任务时不创建线程，空闲线程存活时间默认60S
        ExecutorService threadPool = Executors.newCachedThreadPool();

        Provider p1 = new Provider(queue);
        Provider p2 = new Provider(queue);
        Provider p3 = new Provider(queue);

        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);

        threadPool.execute(p1);
        threadPool.execute(p2);
        threadPool.execute(p3);
        threadPool.execute(c1);
        threadPool.execute(c2);
        threadPool.execute(c3);
        try {
            Thread.sleep(3000);
            p1.stop();
            p2.stop();
            p3.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        threadPool.shutdown();
    }
}
==============================================
package com.wjb.demo.provider_consumer;

/**
 * Created by wjb on 2018/1/11.
 */
public class Data {
    private int id;
    private String name;
    public Data(int id,String name){
        this.id=id;
        this.name=name;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}
===========================================
package com.wjb.demo.provider_consumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by wjb on 2018/1/11.
 */
public class Provider implements Runnable{
    private BlockingQueue queue;
    private volatile boolean isRunning = true;
    private static AtomicInteger count =  new AtomicInteger();

    public Provider(BlockingQueue queue){
        this.queue=queue;
    }

    @Override
    public void run() {
        while (isRunning){
            try {
                Thread.sleep(1000);
                int id = count.incrementAndGet();
                Data data = new Data(id, Integer.toString(id));
                System.out.println("当前线程："+Thread.currentThread().getName()+"生产数据ID是："+id);
                if (!this.queue.offer(data,2, TimeUnit.SECONDS)){
                    System.out.println("数据放入队列失败");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void stop(){
        this.isRunning=false;
    }
}
===============================================
package com.wjb.demo.provider_consumer;

import java.util.concurrent.BlockingQueue;

/**
 * Created by wjb on 2018/1/11.
 */
public class Consumer implements Runnable{
    private BlockingQueue<Data> queue;
    private volatile boolean isRunning = true;
    public Consumer(BlockingQueue queue){
        this.queue=queue;
    }

    @Override
    public void run() {
        while (isRunning){
            try {
                Data data = this.queue.take();
                Thread.sleep(1000);
                System.out.println("消费的数据ID是："+data.getId());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Executor框架


使用有界队列示例demo

package com.wjb.demo.threadpoolexecutor;

/**
 * Created by wjb on 2018/1/11.
 */
public class Task implements Runnable {
    private int id;
    private String name;

    public Task(int id, String name) {
        this.id = id;
        this.name = name;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(5000);
            System.out.println(Thread.currentThread().getName()+"执行任务ID"+this.id);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
=============================================
package com.wjb.demo.threadpoolexecutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by wjb on 2018/1/11.
 * 自定义线程池,有界队列的使用策略
 */
public class ThreadPoolExecutorDemo {
    public static void main(String[] args) {

        /**
         * 首先任务1进来被一个线程执行，任务2，3，4进来时会暂存到队列里去，任务5进来时，
         * 队列已无法暂存，如果当前线程数小于最大线程数，则创建新线程执行此任务。
         * 所以任务5进来时会新建一个线程执行。
如果还有任务6进来的话，此时会执行拒绝策略，JDK默认是AbortPolicy
         */
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2, 60, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(3));

        Task task1 = new Task(1, "任务1");
        Task task2 = new Task(2, "任务2");
        Task task3 = new Task(3, "任务3");
        Task task4 = new Task(4, "任务4");
        Task task5 = new Task(5, "任务5");

        pool.execute(task1);
        pool.execute(task2);
        pool.execute(task3);
        pool.execute(task4);
        pool.execute(task5);

        pool.shutdown();
    }
}
============================================
package com.wjb.demo.threadpoolexecutor;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Created by wjb on 2018/1/11.
 * 自定义拒绝策略，这里可以自行处理，真实场景中，数据是不能丢失的，所以放在缓存中或是其它方式。
 */
public class MyRejected implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println(r.toString());
    }
}
=============================================
执行结果：
Task{id=6, name='任务6'}
pool-1-thread-1执行任务ID1
pool-1-thread-2执行任务ID5
pool-1-thread-1执行任务ID3
pool-1-thread-2执行任务ID2
pool-1-thread-1执行任务ID4

使用无界队列示例DEMO

package com.wjb.demo.threadpoolexecutor;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by wjb on 2018/1/11.
 * 线程池 使用无界队列的策略
 */
public class ThreadPoolExecutor2 implements Runnable{
    private int id;
    private String name;
    private static AtomicInteger count = new AtomicInteger();
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    @Override
    public void run() {
        try {
            int id = count.incrementAndGet();
            System.out.println(Thread.currentThread().getName()+"执行的任务ID："+id);
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();

        /**
         * 来一个任务新建一个线程，当线程到corePoolSize时不再创建线程，所以这里的最大线程数一般和核心线程一样。所有新来的任务都暂存到队列中
         * 无界队列不会拒绝新来的任务直到内存耗净。
         */
        ExecutorService executor = new ThreadPoolExecutor(5,5,120, TimeUnit.SECONDS,queue);

        for (int i = 1;i<= 20;i++){
            executor.execute(new ThreadPoolExecutor2());
        }
        try {
            Thread.sleep(1000);
            System.out.println("队列大小"+queue.size());
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executor.shutdown();
    }
}
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,921评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 87,635评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,393评论 0赞 338
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,836评论 1赞 277
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,833评论 5赞 368
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,685评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,043评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,694评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 42,671评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,670评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,779评论 1赞 332
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,424评论 4赞 321
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,027评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,984评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,214评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,108评论 2赞 351
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,517评论 2赞 343

推荐阅读更多精彩内容