快捷導(dǎo)航

java管道piped輸入流與輸出流應(yīng)用場景案例分析

更新時間：2022年02月21日 11:11:00 作者：kl

這篇文章主要介紹了java管道流PipedInputStream與PipedOutputStream(輸入流與輸出流)的應(yīng)用場景案例分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步

前言

PipedInputStream 和 PipedOutputStream 設(shè)計用來解決跨線程的字節(jié)數(shù)據(jù)傳輸。它們總是成對出現(xiàn)的，而在使用上，也只能工作在兩個不同的線程上，在一個線程里使用管道輸入和輸出流可能會造成死鎖。網(wǎng)上有很多介紹這兩個存在于 io 包下的 api。卻幾乎找不到一個寫 PipedInputStream 的使用場景的，所以本文結(jié)合實際業(yè)務(wù)，來聊一聊 PipedInputStream 的應(yīng)用。

原理簡介

我們知道，輸出流寫數(shù)據(jù)，輸入流讀數(shù)據(jù)，PipedInputStream 和 PipedOutputStream 也一樣，在 PipedOutputStream 的內(nèi)部有一個 PipedInputStream 類型的 sink屬性，用來接收 PipedOutputStream 寫入的字節(jié)數(shù)據(jù)。

而在 PipedInputStream 內(nèi)部，定義了一個默認(rèn)為 1024 大小的字節(jié)數(shù)組 buffer，作為數(shù)據(jù)傳輸?shù)木彌_區(qū)。這樣一來，就變成了 PipedOutputStream 往 buffer 里寫數(shù)據(jù)，當(dāng)寫滿了 buffer 時，便使用 notifyAll() 喚醒讀數(shù)據(jù)的線程可以讀數(shù)據(jù)了，然后阻塞 1s 后繼續(xù)嘗試寫數(shù)據(jù)。

PipedInputStream 從 buffer 里讀數(shù)據(jù)，當(dāng)數(shù)據(jù)讀完 buffer 為空時，便 notifyAll() 喚醒寫的線程可以寫數(shù)據(jù)了，然后阻塞 1s 后繼續(xù)嘗試讀數(shù)據(jù)。

PipedOutputStream 端數(shù)據(jù)寫完后，調(diào)用 close() 方法，會標(biāo)記 PipedInputStream 里的 closedByWriter=true。此時，從 buffer 讀取數(shù)據(jù)，會返回 -1。標(biāo)識了數(shù)據(jù)讀完到達(dá)了流的末尾了。

使用場景概述

public static void main(String[] args) {
        try (PipedOutputStream out = new PipedOutputStream();
             PipedInputStream in = new PipedInputStream(out)) {
            new Thread(() -> {
                try {
                    out.write("hello kl".getBytes(StandardCharsets.UTF_8));
                    out.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            int receive;
            while ((receive = in.read()) != -1) {
                System.err.print((char) receive);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

上面代碼演示了，在一個線程里寫數(shù)據(jù)，然后在 main 線程讀數(shù)據(jù)的場景，完成了跨線程的數(shù)據(jù)傳輸。寫到這里，都挺干巴巴的，很多人看了后肯定也不知道它到底能干啥，有啥作用，繼續(xù)往下看。

實際應(yīng)用

簡單的理解了原理后，寫了一個簡單的演示 demo，但是 demo 不能說明啥問題，那從一個線程傳輸字節(jié)到另一個線程到底有啥用呢？博主，簡單的的總結(jié)下:通過 java 應(yīng)用生成文件，然后需要將文件上傳到云端的場景，都可以用管道流。相同的業(yè)務(wù)場景，在沒了解管道流之前，都是先將文件寫入到本地磁盤，然后從文件磁盤讀出來上傳到云盤。了解這個后，可以腦補(bǔ)出很多的業(yè)務(wù)場景了(真實業(yè)務(wù)場景，都是博主遇到過的)，比如：

案例一：EXCEL 文件導(dǎo)出功能

之前有一個文件導(dǎo)出的功能，但是因為，導(dǎo)出的文件比較大，導(dǎo)出下載完的時間非常長，所以，設(shè)計成了，頁面點擊導(dǎo)出后，后臺觸發(fā)導(dǎo)出任務(wù)，然后將mysql 中的數(shù)據(jù)根據(jù)導(dǎo)出條件查詢出來，生成 Excel文件，然后將文件上傳到 oss，最后像觸發(fā)導(dǎo)出任務(wù)的人的釘釘發(fā)一個下載文件的鏈接。之前的做法，正如上面所言，先將文件寫到本地，然后從本地目錄讀出來上傳到 oss，下面演示下管道流一步到位的方式：

public static void main(String[] args) {
        try (PipedOutputStream out = new PipedOutputStream();
             PipedInputStream in = new PipedInputStream(out)) {
            new Thread(() -> {
                Listdatabase = new LinkedList<>();
                try {
                    //文件生成
                    ExcelUtils.getInstance().exportObjects2Excel(database,out);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
            //文件上傳
            ossClient.putObject("test","test.xlsx",in);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

案例二：XML 文件數(shù)據(jù)傳輸

此類需求常見于和銀行以及金融機(jī)構(gòu)對接時，要求上報一些 xml 格式的數(shù)據(jù)，給到指定的 ftp、或是 oss 的某個目錄下，用于對賬。其實從文件上傳的場景來說，和上面的案例一是一樣。也是我總結(jié)的那樣，在內(nèi)存里生成文件，然后上傳到云端，偽代碼如下：

public static void main(String[] args) {
        try (PipedOutputStream out = new PipedOutputStream();
             PipedInputStream in = new PipedInputStream(out)) {
            new Thread(() -> {
                Listdatabase = new LinkedList<>();
                try(GZIPOutputStream gzipOut = new GZIPOutputStream(out)) {
                    Marshaller marshaller = JAXBContext.newInstance(Object.class).createMarshaller();
                    marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, Boolean.TRUE);
                    marshaller.marshal(database,gzipOut);
                } catch (IOException | JAXBException e) {
                    e.printStackTrace();
                }
            }).start();
            //文件上傳
            ossClient.putObject("test","test.xml.gz",in);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

總體來說，和案例一沒啥區(qū)別，只是案例二多了一步壓縮操作，最終上傳的文件是一個 gzip 的壓縮包，壓縮包內(nèi)是 xml 文件。用這種方式可以大大減少文件的體積、提升上傳的速度