Java进阶之并发初探

一、什么是并发

每个进程都有自己独立的进程空间，编写进程的并发， 会频繁发生上下文切换

 

二、并发实现

我们用不同线程实现的任务都应该去实现Runable接口，重写run()方法，比如我们创建一个用于倒计时的LiftOff类：

public class LiftOff implements Runnable {
    private static int taskCount = 0;
    private final int id = taskCount++;
    private int countDown = 10;
    public LiftOff() {}
    public LiftOff(int countDown) { this.countDown = countDown; }
    private String status() {return "#" + id + "(" + (countDown > 0 ? countDown : "LiftOff") + ") ";}
    public void run() {
        while (countDown-- > 0) {
            System.out.print(status());
            Thread.yield();
        }
    }
}

注意这里的taskCount是static的，因为我们想要对每个线程都赋予不同的id，所以应该创建一个静态变量，java中的静态变量存储在虚拟机的方法区中，为所有线程共享。

id是final的，因为一旦创建了就不会去改变它。

Thread.yield() 表示告诉线程调度机制：你的工作已经做的差不多了，可以让别的线程使用CPU了，不过也只是个暗示，不一定表示会被采纳。

 

1.Thread类

另外我们实现一个线程也可以采用继承thread类的形式，只需要重写run()方法就可以。

 

2.使用Executor

Executor相当与提供了一个对Thread类的管理，例子：

public class CachedThreadPool {
    public static void main(String[] args) {
        ExecutorService exec =  Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            exec.execute(new LiftOff());
        }
        exec.shutdown();

    }
}

 

来看一下Exector提供的线程池都有哪些呢？

newCachedThreadPool	通常用来执行一些生存期很短的异步任务
newFixedThreadPool
newScheduledThreadPool
newSingleThreadExecutor	连续运行额的任务（长期存活的任务）

 

3.从任务中返回值

实现Callable接口而不是Runable，写一个Fibonacci类能够返回长度为n的序列，使用submmit和get取得线程执行的返回值

public class MultiFibonacci implements Callable<String> {
    private  int n = 0;
    private Fibonacci fibonacci;
    private StringBuilder seq;
    public MultiFibonacci(int n) {
        this.n = n;
        fibonacci = new Fibonacci();
        seq = new StringBuilder();
    }
    public String call() {
        for (int i = 0 ; i < n; i++) {
            seq.append(fibonacci.next());
            seq.append(" ");
        }
        return seq.toString();
    }
    public void run() {
        for (int i = 0 ; i < n; i++) {
            seq.append(fibonacci.next());
            seq.append(" ");
        }
        System.out.println(seq);
    }
    public static void main(String[] agrs) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        List<Future<String>> list = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            list.add(executorService.submit(new MultiFibonacci(i)));
        }
        for (Future<String> res : list) {
            try {
                System.out.println(res.get());
            } catch (InterruptedException e) {
                System.out.println(e);
                return;
            } catch (ExecutionException e) {
                System.out.println(e);
                return;
            } finally {
                executorService.shutdown();
            }
        }
    }
}

 

4.后台线程

在线程启动之前，使用.setDaemon() 将它设置为后台线程

 

5.加入线程

某个线程在另一个线程t上调用t.join()，此线程将被挂起，直到目标线程执行结束。

 

二、资源共享

并发中常见的问题就是不同的线程访问同一资源造成冲突。

 

1.synchronized关键字

对于要访问的共享资源，首先应该将它封装进一个对象，然后把所有要访问这个对象资源的方法标记为synchronized。

如果某个对象的方法g和f需要访问共享的资源，则可以将方法声明为：

synchronized void f();

synchronized void g();

对于同一对象而言，其所有synchronized方法都共享同一个锁。即为对象锁。所以当调用该对象上任一synchronized方法时，都被加锁，这时如果要调用该对象其他的synchronized方法，只有等前一个方法调用完毕并释放了锁之后才能被调用，即此时该对象的所有其他synchronized方法都被阻塞。即其它线程对该对象所有同步代码部分的访问都被暂时阻塞，即加锁是加在对象上的。

 

=> 有时候会碰见在方法前加synchronized和在代码块前加synchronized，这两者有什么不同呢？

1）.同步方法 

    即有synchronized关键字修饰的方法。 

    由于java的每个对象都有一个内置锁，当用此关键字修饰方法时， 

    内置锁会保护整个方法。在调用该方法前，需要获得内置锁，否则就处于阻塞状态。

     注： synchronized关键字也可以修饰静态方法，此时如果调用该静态方法，将会锁住整个类

 

2）.同步代码块 

    即有synchronized关键字修饰的语句块。 

    被该关键字修饰的语句块会自动被加上内置锁，从而实现同步

 

 

 

 

 

 

注意： 同步方法锁的范围比较大，而同步代码块范围要小点，因为一般同步的范围越大，性能就越差，一般需要加锁进行同步的时候，肯定是范围越小越好，这样性能更好。

同步是一种高开销的操作，因此应该尽量减少同步的内容。通常没有必要同步整个方法，使用synchronized代码块同步关键代码即可。

 

 

 注：实际开发中很少直接用synchronized关键字,因为它会导致性能很慢，尽管控制了线程同步，但是并发量极小，每次只能保证一个线程的访问，无法做到细粒度的控制，而且只适用于单点的情况（无法适用于集群）；因此实际开发中往往会使用redis分布式锁。

 

2. volatile 关键字

• 使用volatile关键字，保证变量可见性（直接从内存读，而不是从线程cache读）

记住，加锁机制可以确保可见性和原子性，而volatile只能确保可见性 

 

 

2.使用显示的Lock对象

 

三、终结任务

来看一个并发的实例：

写一个实时统计公园几个大门口通过的人数的程序，任意一个门口人数增加时，就表示进入公园的总人数增加

class Count {
    private int count = 0;
    private Random random = new Random(47);
    public synchronized int increment() {
        return ++count;
    }
    public synchronized int value() {return count;}
}

class Entrance implements Runnable {
    private int number;
    private static Count count = new Count();
    private static List<Entrance> entrances = new ArrayList<>();
    private static boolean isCancled;
    private int id;
    public Entrance(int id) {
        this.id = id;
        entrances.add(this);
    }
    @Override
    public void run() {
        while (!isCancled) {
            ++number;
            System.out.println(this + " total: " + count.increment());
        }
        try {
            TimeUnit.MICROSECONDS.sleep(100);
        } catch (InterruptedException e) {
            System.out.println("sleep interrupted!");
        }
        System.out.println("stooping " + this);
    }
    public static void cancle() {isCancled = true;}
    public synchronized int getValue() {return number;}
    public static int getTotalSum() {
        int sum = 0;
        for (Entrance entrance : entrances) {
            sum += entrance.getValue();
        }
        return sum;
    }
    public static int getTotalCount() { return count.value(); }
    public String toString() { return "Entrance" + id + ":" + getValue(); }
}

public class OrnamentalGarden {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            service.execute(new Entrance(i));
        }
        TimeUnit.SECONDS.sleep(1);
        Entrance.cancle();
        service.shutdown();
        if(!service.awaitTermination(250, TimeUnit.MILLISECONDS))
            System.out.println("some is not terminated");
        System.out.println("Total: " + Entrance.getTotalCount());
        System.out.println("Sum: " + Entrance.getTotalSum());
    }
}

 

 

四、自己动手写一个线程池

 一个线程池，就是一定数量的线程队列，每次都取一个线程来执行任务，当线程池中的线程不够时，当前任务就必须等待，否则就取出一个线程来执行该任务。想一想这个模型是不是就是生产者-消费者模型 ！

数据结构可以采用LinkedBlockingQueue，LinkedBlockingQueue实现是线程安全的，实现了先进先出等特性，是作为生产者消费者的首选

另外需要注意的就是需要对queue进行同步，使用synchronized关键字来同步代码块的方式！使用到queue的代码块用synchronized包围起来

import java.util.concurrent.LinkedBlockingQueue;

//线程池实现，其实就是一个生产者消费者模型

class Task implements Runnable{
    private int num;

    public Task(int num) {
        this.num = num;
    }
    public void run() {
        System.out.println("Task" + num + " is running");
    }
}

public class ThreadPool {
    private final LinkedBlockingQueue<Runnable> queue;
    private final int nThreads;
    private final PoolWorker[] threads;

    public ThreadPool(int nThreads) {
        this.nThreads = nThreads;
        queue = new LinkedBlockingQueue<>();
        threads = new PoolWorker[nThreads];

        for (int i = 0; i < nThreads; i++) {
            threads[i] = new PoolWorker();
            threads[i].start();
        }
    }

    private class PoolWorker extends Thread {
        public void run() {
            Runnable task;
            while (true) {
                synchronized (queue) {
                    while (queue.isEmpty()) {
                        try {
                            queue.wait();
                        } catch (InterruptedException e) {
                            System.out.println(e);
                        }
                    }
                    task = queue.poll();
                }
                try {
                    task.run();
                } catch (RuntimeException e) {
                    System.out.println(e);
                }
            }
        }
    }

    public void execute(Runnable task) {
        synchronized (queue) {
            queue.add(task);
            queue.notify();
        }
    }

    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(8);
        for (int i  = 0; i < 5; i++) {
            Task task = new Task(i);
            threadPool.execute(task);
        }
    }
}