Java多线程与阻塞队列

目录

    ※什么是线程

        ▲进程

        ▲线程

    ※Java多线程

        ▲多线程的实现

        ▲线程的种类

        ▲Runnable实现多线程

    ※阻塞队列

        ▲LinkedBlockingQueue与ArrayBlockingQueue

        ▲使用场景

        ▲阻塞方法

※什么是线程

要了解什么是线程。需要先明白什么是进程。那么什么是进程？

▲进程

      系统中正在运行中的程序就是一个进程，进程是程序执行的一个实例。进程是系统进行资源分配的独立实体，每个进程都拥有其独立的地址空间。一般情况下，进程无法访问另一个进程的变量和数据结构，如果想要访问另一个进程的资源，需要进程间通信，如借助管道，文件，套接字等。

▲线程

      了解了进程，线程就比较好解释了。线程就是进程中的实际运作单位，一个进程中可以并发多个线程，每条线程并行执行不同的任务。就比如打开Chrome后，你可以同时开启多个标签页，一边放视频，一边查资料，Chrome是一个运行中的程序，也就是进程，每个标签页是一个线程。线程是操作系统能够运算调度的最小单位。

※Java多线程

对线程概念有了解后来看一下Java里的多线程。

▲多线程的实现

在Java中实现多线程有三种方式： 

    1.继承Thread类 

    2.实现Runnable接口 

    3.使用Callable和Future（没用过不了解～） 

相较于继承Thread，更推荐去实现Runnable接口，

因为： 

    ◆ 实现接口可以避免Java中对类的单继承的限制 

    ◆ 更适合使用多个相同程序到吗的线程去处理同一个资源

▲线程的种类

       Java中的线程有两类，一类是用户线程；一类是守护线程，守护线程是一类特殊的线程，主要用做程序中后台调度与支持性工作。 

       程序的入口main方法就是典型的用户线程，GC（垃圾回收器）是典型的守护线程。所以JVM在启动后本身就一定以多线程的状态运行（main与GC）。

▲Runnable实现多线程

       用一小段程序来使用Runnable实现多线程。刚巧，我这里有一本TXT格式的电子书《丧钟为谁而鸣》，我希望获取到它的所有行的内容，并通过多个线程将每一行打印在Console中。关于电子书，可查阅这篇博文。

public class thread {
    private static boolean flag = true;
    private static AtomicInteger count = new AtomicInteger(0);
    public static void main(String[] args) {
        List txtList = readFile("G:\\下载\\丧钟为谁而鸣.txt");
        BlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();
        ExecutorService executor = Executors.newCachedThreadPool();
        executor.execute(new ThreadAddTextToQueue(queue,10,txtList));
        executor.execute(new ThreadDealTxt(queue));
        executor.execute(new ThreadDealTxt(queue));
        executor.execute(new ThreadDealTxt(queue));
        executor.execute(new ThreadDealTxt(queue));
        executor.shutdown();
    }
/**从这段main方法中能够得知什么呢？“读取”小说中的每一行并将其放在一个线程非安全的List中，
启用一个线程任务将线程非安全的List中的数据批量的放到线程安全的BlockingQueue队列中。
另启用了四个线程任务处理队列中的数据并打印在Console中。*/

/**使用实现Runnable接口的方式创建一个线程ThreadAddTextToQueue，
将线程非安全的List中的文本数据批量获取并添加到阻塞队列BlockingQueue中，
并在队列尾添加一个结束标志“close”。*/
      static class ThreadAddTextToQueue implements Runnable {
        private final BlockingQueue<Object> queueList;
        private Integer size = null;
        private List<String> listAll = null;
        public ThreadAddTextToQueue(BlockingQueue queueList,Integer size,List<String>listAll){
            this.queueList = queueList;
            this.size = size;
            this.listAll = listAll;
        }
         public void run() {
             if (listAll != null && listAll.size() > 0 && size != null && size > 0) {
                 Integer length = listAll.size() / size.intValue();
                 for (int i = 0; i <= length; i++) {
                     if ((i + 1) * size < listAll.size()) {
                         queueList.addAll(listAll.subList(i * size, (i + 1) * size));
                     } else {
                         queueList.addAll(listAll.subList(i * size, listAll.size()));
                     }
                 }
             }
             try {
                 queueList.put("close");
             } catch (InterruptedException e) {
                e.printStackTrace();
             }
         }
     }

     static class ThreadDealTxt implements Runnable{
        private BlockingQueue<Object> queueList;
        public ThreadDealTxt(BlockingQueue queueList){
            this.queueList = queueList;
        }
         public void run() {
            while(flag){
                if(!queueList.isEmpty()){
                    try{
                        Object takeOne = queueList.take();
                        if("close".equals(takeOne+"")){
                            flag = false;
                            break;
                        }
                        dealTake((String)takeOne);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
         }
     }

    public static void dealTake(String txt){
        System.out.println("---" + count.incrementAndGet()+ "---" + txt);
    }

    public static  List<String> readFile(String filePath){
        List<String> textList = new LinkedList<String>();

        try {
            //BufferedReader file = new BufferedReader(new FileReader(filePath));
            BufferedReader file = new BufferedReader(new InputStreamReader(new FileInputStream(filePath),"UTF-8"));
            String line = file.readLine();
            while(null != line){
                textList.add(line);
                line = file.readLine();
            }
            file.close();
            return textList;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return textList;
    }
}

       众所周知，一提到多线程，并发就涉及到线程同步，内存一致性问题，在这里，为表现出有别于一般的单线程开发环境，‘多此一举’的使用ThreadAddTextToQueue 这样一个线程去进行这样的转换处理过程。

       实现Runnable接口只需实现接口方法run(),ThreadDealTxt 线程使用同样的程序代码来处理同一份数据。

       在读取文件时，我写了两行代码：

   //BufferedReader file = new BufferedReader(new FileReader(filePath));
   BufferedReader file = new BufferedReader(new InputStreamReader(new FileInputStream(filePath),"UTF-8"));

       我为什么采用了第二种方式读取而放弃了第一种方式呢？在解释这个疑问之前，我们可以先看一下FileReader的源码：

  public class FileReader extends InputStreamReader {

   /**
    * Creates a new FileReader, given the name of the
    * file to read from.
    *
    * @param fileName the name of the file to read from
    * @exception  FileNotFoundException  if the named file does not exist,
    *                   is a directory rather than a regular file,
    *                   or for some other reason cannot be opened for
    *                   reading.
    */
    public FileReader(String fileName) throws FileNotFoundException {
        super(new FileInputStream(fileName));
    }    
  }

       可以发现，FileReader是继承自InputStreamReader，构造时本质上仍然是调用父类的构造方法。而FileReader在屏蔽了这些细节的同时，却有可能存在编码问题，导致读取的文件内容是乱码，其原因是读取文件时采用IDEA中设置的文件编码格式或系统默认的编码格式与文件编码格式不同。
所以为避免乱码问题，可以直接使用InputStreamReader流并显式的设置文件解码格式。

    public InputStreamReader(InputStream in, String charsetName)
        throws UnsupportedEncodingException
    {
        super(in);
        if (charsetName == null)
            throw new NullPointerException("charsetName");
        sd = StreamDecoder.forInputStreamReader(in, this, charsetName);
    }

※阻塞队列

在处理文件每行的数据时使用了阻塞队列LinkedBlockingQueue。

▲LinkedBlockingQueue与ArrayBlockingQueue

LinkedBlockingQueue是java.util.concurrent包下专为多线程同步提供的线程安全队列，与之相似的还有ArrayBlockingQueue，它们都实现BlockingQueue接口，都是阻塞队列； 

区别在于，

    1.底层实现不同：前者底层使用链表，后者使用Object数组。 

    2.锁的机制不同：LinkedBlockingQueue的锁是分离的，生产者有putLock，消费者有takeLock;ArrayBlockingQueue只用一个锁。其原因在于就二者的底层实现不同，Object数组的形式使得无法对ArrayBlockingQueue同时进行take与put操作，它必须准确的知晓数组开始与结束的位置；

  /** Main lock guarding all access */
  final ReentrantLock lock;

LinkedBlockingQueue由于其链表的结构，不需要知道这些，让GC去操心那些节点吧。

/**Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

     3.计数方式不同： 基于同样的原因，ArrayBlockingQueue与LinkedBlockingQueue的计数实现也不同。ArrayBlockingQueue维护基础数据类型int来记录队列长度

/** Number of elements in the queue */
    int count;

LinkedBlockingQueue则需要使用java.util.concurrent.aotmic.AtomicInteger这样的线程安全的原子类来确保数据的正确性

/** Current number of elements */
    private final AtomicInteger count = new AtomicInteger();

    4.构造方式不同：ArrayBlockingQueue的所有构造方法都必须提供初始化容量参数，LinkedBlockingQueue则允许无参构造，容量默认为Integer.MAX_VALUE。

▲使用场景

        在示例中，我们使用一个线程入队，多个线程出队，这是典型的生产者消费者模型的应用场景，而阻塞队列就是生产者存放、消费者来获取的容器。

▲阻塞方法

 阻塞队列需要支持阻塞的插入和移除方法：
    ★ 阻塞的插入方法：队列满时，队列阻塞试图插入元素的线程，直至队列不满。 

    ★ 阻塞的移除方法：队列空时，队列阻塞试图获取元素的线程，直至队列非空。

具体到示例代码中，

    尝试将一行文字加入队列中使用阻塞方法put();

    尝试从队列里移除一条数据使用阻塞方法take()。

void put(E e) throws InterruptedException;

E take() throws InterruptedException;

此外，从阻塞队列LinkedBlockingQueue中出队，除使用阻塞方法take()，

还有两种方式： 

    ★ remove()出队，若队列为空，则会抛出异常； 

    ★ poll()出队，若队列为空，则会返回null。