快捷導(dǎo)航

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

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

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

前言

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

原理簡(jiǎn)介

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

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

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

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

使用場(chǎng)景概述

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();
        }
    }

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

實(shí)際應(yīng)用

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

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

之前有一個(gè)文件導(dǎo)出的功能，但是因?yàn)?，?dǎo)出的文件比較大，導(dǎo)出下載完的時(shí)間非常長，所以，設(shè)計(jì)成了，頁面點(diǎn)擊導(dǎo)出后，后臺(tái)觸發(fā)導(dǎo)出任務(wù)，然后將mysql 中的數(shù)據(jù)根據(jù)導(dǎo)出條件查詢出來，生成 Excel文件，然后將文件上傳到 oss，最后像觸發(fā)導(dǎo)出任務(wù)的人的釘釘發(fā)一個(gè)下載文件的鏈接。之前的做法，正如上面所言，先將文件寫到本地，然后從本地目錄讀出來上傳到 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)對(duì)接時(shí)，要求上報(bào)一些 xml 格式的數(shù)據(jù)，給到指定的 ftp、或是 oss 的某個(gè)目錄下，用于對(duì)賬。其實(shí)從文件上傳的場(chǎng)景來說，和上面的案例一是一樣。也是我總結(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ū)別，只是案例二多了一步壓縮操作，最終上傳的文件是一個(gè) gzip 的壓縮包，壓縮包內(nèi)是 xml 文件。用這種方式可以大大減少文件的體積、提升上傳的速度