Java 異步處理簡單實踐

同步與異步

       一般同步意味着一個任務的某個處理過程會對多個線程在用串行化處理，而異步則意味着某個處理過程能夠容許多個線程同時處理。

       異步一般表明着更好的性能，由於它很大程度上依賴於緩衝，是典型的使用空間換時間的作法，例如在計算機當中，高速緩存做爲cpu和磁盤io之間的緩衝地帶協調cpu高速計算能力和磁盤的低速讀寫能力。

volatile

       應用場景：檢查一個應用執行關閉或中斷狀態。由於此關鍵字拒絕了虛擬對一個變量屢次賦值時的優化從而保證了虛擬機必定會檢查被該關鍵字修飾的變量的狀態變化。

CountDownLatch

       應用場景：控制在一組線程操做執行完成以前當前線程一直處於等待。例如在主線程中執行await()方法阻塞主線程，在工做線程執行完邏輯後執行countDown()方法。

本文示例場景：

       1，從控制檯發送消息到消息服務器(由一個隊列模擬)。

       2，將消息隊列寫入到文件(對寫文件的操做設置延時以模擬性能瓶頸)。

       3，消息服務器做爲控制檯和文件寫入之間的緩衝區。

 

示例代碼：

      注：往消息隊列添加消息能夠經過for循環一次性加入，本文爲了便於觀察文件和隊列的變化而採用了控制檯輸入，實際寫一行文件記錄速度應該高於手速，因此本文示例中增長了線程sleep時間。

package org.wit.ff.ch2;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 
 * <pre>
 * 簡單異步處理示例.
 * </pre>
 * 
 * @author F.Fang
 * @version $Id: AsyncHandler.java, v 0.1 2014年10月23日 下午11:37:54 F.Fang Exp $
 */
public class AsyncHandler {

    /**
     * 控制資源釋放.
     */
    private CountDownLatch latch;

    /**
     * 處理完成標識.
     */
    private volatile boolean handleFinish;

    /**
     * 消息寫入本地文件完成.
     */
    private volatile boolean sendFinish;

    /**
     * 阻塞隊列.
     */
    private BlockingQueue<String> queue;

    private BufferedWriter bw;

    public AsyncHandler(CountDownLatch latch) {
        this.latch = latch;
        /**
         * 使用鏈表實現.
         */
        queue = new LinkedBlockingQueue<String>();
        File file = new File("E:/hello.txt");
        try {
            bw = new BufferedWriter(new FileWriter(file));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void handle() {
        // 模擬性能瓶頸的執行過程,3s處理一條消息.
        new Thread() {
            public void run() {
                while (!handleFinish) {
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e1) {
                        // 不作處理.
                    }
                    String s = queue.peek();
                    if (s != null) {
                        queue.poll();
                        try {
                            bw.write(s);
                            bw.newLine();
                        } catch (IOException e) {
                        }
                    }
                    // 若隊列爲空而且消息發送完成.
                    if (queue.isEmpty() && sendFinish) {
                        // 計數器1->0
                        latch.countDown();
                        // 讓處理過程結束.
                        handleFinish = true;
                        break;
                    }
                }
            }

        }.start();

    }

    /**
     * 
     * <pre>
     * 給出消息發送完成的標識.
     * </pre>
     *
     */
    public void sendFinish() {
        sendFinish = true;
    }

    /**
     * 
     * <pre>
     * 資源釋放.
     * </pre>
     *
     */
    public void release() {
        System.out.println("release!");
        if (bw != null) {
            try {
                bw.close();
            } catch (IOException e) {
                // TODO 打印日誌.
            }
        }
        //其實使用queue = null就夠了.
        if (queue != null) {
            queue.clear();
            queue = null;
        }
    }

    /**
     * 
     * <pre>
     * 往隊列發送消息.
     * </pre>
     *
     * @param text
     */
    public void sendMsg(String text) {
        if (text != null && !text.isEmpty()) {
            queue.add(text);
        }
    }

    public static void main(String[] args) {
        CountDownLatch latch = new CountDownLatch(1);
        AsyncHandler handler = new AsyncHandler(latch);
        handler.handle();

        // 作一次檢查.
        Scanner scanner = new Scanner(System.in);
        while (true) {
            String text = scanner.next();
            // 若用戶選擇退出.
            if ("exit".equals(text)) {
                // 表示消息已經發送完成.
                handler.sendFinish();
                break;
            }
            handler.sendMsg(text);
        }

        try {
            // 阻塞主線程等待消息寫入到本地文件完成.
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 釋放資源 文件流,隊列.
        handler.release();
        // 關閉控制檯輸入.
        scanner.close();
    }

}