從文件導數據到數據庫的性能優化思路（筆記）

概述

     最近公司一.NET項目需要對其日志Log入數據庫統計，寫這個腳本導入的任務便落到我身上了。采用了熟練的Java,這個日志也不是很大，一個文件大概幾兆，有上萬條數據，一天大概有7,8個文件需要這樣的導入處理。由於之前寫Web沒有這么去批處理這么多數據，所以沒有太注意性能，第一個版本程序導入速度慢的嚇人，一個文件導完可能需要10多分鍾，也就是說如果把每天的文件導完可能需要2個多小時的時間，聽聽就很蛋疼，最終經過優化后，一個文件導入也就幾秒，甚至可以更短。目標日志文件的信息都是按行存儲，所以程序中按行讀取后，然后進行相應的字符串截取入庫。下面則為思路分享以及主要代碼的分享。

 

優化思路

     1.程序流程：

     　　程序先讀取本地的文件到內存，然后把內存的數據批量Insert到數據庫。

     2.歸納：

　　　  可以看出首先程序需要進行文件IO操作，然后則是數據JDBC操作，所以優化方向大致可以是以下幾個：

          a.文件IO優化

          b.JDBC操作優化

          c.使用多線程並行JDBC操作

 文件常見IO簡介

     Java的文件讀寫操作大概有這么幾種方式，但是我們應該注意幾種文件操作方式的區別，哪些操作方式適合不同的數據文件對象。

     1.（InputStream/OutputStream)    為字節輸入/輸出流，這種讀寫方式都是按一定字節量讀取數據。

     2. （FileInputStream/FileOutputStream) 此方法繼承自上面的（InputStream/OutpustStream）,同樣按字節流輸入/輸出，用於讀取圖像之類的原始字節流

     3.（FileReader/FileWriter) 此方法適用於按字符流的文件操作

     4. （BufferedReader/BufferedWriter) 從字符輸入流中讀取文本，緩沖各個字符，從而實現字符、數組和行的高效讀取。

 

      注：更詳細的IO操作說明，請查看具體的JDK文檔。

　　 此處我采用的BufferedReader按行讀取， 代碼片段：

    public static List<String> getLogLinesByBuf(String filePath){
        
        List<String> items = new ArrayList<String>();
        File file = new File(filePath);    
        BufferedReader reader;
        if (file.exists()) {
            
            try {
                reader = new BufferedReader(new FileReader(file));
                String temp = "";
                while((temp = reader.readLine()) != null) {
                    items.add(temp);
                }            
                //close
                reader.close();
            } catch (Exception e) {                
                e.printStackTrace();
            }
        } else {
            System.out.println("該路徑文件不存在.");
        }        
        return items;
    }

 

PreparedStatement和Statement 

        JDBC操作我們經常會用到PreparedStatement和Statement,PreparedStatement相對Statement來講，PreparedStatement擁有預編譯能力，性能更好，2者其它的優缺點比較可以查看相關的資料。另外，平常我們插入數據都是一條，2條，當完成成千上萬條數據插入操作的時候，你會看到性能是直線下降的，所以這里會采用sql批處理。

       代碼片段：

　　　　

    public static void insertLogInfo(List<String> data) {
        
        String sql = "INSERT INTO log_info(date_time,s_sitename,s_ip,cs_method,cs_uri_stem,cs_uri_query,"
                + "s_port,cs_username,c_ip,cs_user_agent,sc_status,sc_substatus,sc_win32_status"
                + ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)";
        Connection conn = DBSource.getConnection();
        int count = 0;
        try {
            conn.setAutoCommit(false);
            PreparedStatement prest = conn.prepareStatement(sql);  
            
            for(String str : data) {
                String[] arr = str.split(" ");
                prest.setString(1, arr[0]+" "+arr[1]);
                prest.setString(2, arr[2]);
                prest.setString(3, arr[3]);
                prest.setString(4, arr[4]);
                prest.setString(5, arr[5]);
                prest.setString(6, arr[6]);
                prest.setString(7, arr[7]);
                prest.setString(8, arr[8]);
                prest.setString(9, arr[9]);
                prest.setString(10, arr[10]);
                prest.setString(11, arr[11]);
                prest.setString(12, arr[12]);
                prest.setString(13, arr[13]);
                //添加到批處理
                prest.addBatch();    
            }

            int [] intarr = prest.executeBatch();
            conn.commit();  
            prest.clearBatch();                      
            prest.close();
            conn.close();
            for (int j = 0 ; j < intarr.length; j++) {
                if (intarr[j] > 0) {
                    count +=1;
                }
            }    
        } catch (Exception e) {
            System.out.println(new Date().toLocaleString()+":數據庫插入操作失敗"+e.getMessage());
        }      
        System.out.println("本次操作成功插入"+count+"行數據");    
    }

 

多線程並行處理

         例如本來1萬條數據是一個線程進行JDBC批量提交，現在啟用5個線程並行處理，每個線程2000條數據，甚至你可以根據數據量來分配更多線程來完成同步提交，性能提升會比較明顯。

       代碼片段：

　　Thread類： 　

package com.xj.dbsource;

import java.io.File;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Date;
import java.util.List;

import com.json.utils.JsonFileUtils;
import com.xj.iislog.bean.JDBCInfo;



/**
 * 
 * @author Ziv
 * 數據操作源
 */
public class DBSource extends Thread {
    
    //聲明對象
    private static Statement statement;
    //連接對象
    private static Connection conn;
    
    private List<String> data;
        
    public DBSource(List<String> data) {
        super();
        this.data = data;
    }

    public void run(){
        System.out.println(System.currentTimeMillis());
        DBSource.insertLogInfo(data);
        System.out.println(System.currentTimeMillis());
    }
    
    /**
     * 
     * @param sql
     * @return int
     */
    @SuppressWarnings("deprecation")
    public int insert(String sql) {
        
        int result = 0;
        try {
            conn = getConnection();
            statement = conn.createStatement();
            result = statement.executeUpdate(sql);        
            //關閉連接
            conn.close();
        } catch (SQLException e) {
            System.out.println(new Date().toLocaleString()+":數據庫插入操作失敗" +e.getMessage());
        }
        
        return result;
    }
    
    /**
     * prepared方式入庫
     * @param arr
     * @return
     * @throws SQLException
     */
    @SuppressWarnings("deprecation")
    public static void insertLogInfo(List<String> data) {
        
        String sql = "INSERT INTO log_info(date_time,s_sitename,s_ip,cs_method,cs_uri_stem,cs_uri_query,"
                + "s_port,cs_username,c_ip,cs_user_agent,sc_status,sc_substatus,sc_win32_status"
                + ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)";
        Connection conn = DBSource.getConnection();
        int count = 0;
        try {
            conn.setAutoCommit(false);
            PreparedStatement prest = conn.prepareStatement(sql);  
            
            for(String str : data) {
                String[] arr = str.split(" ");
                prest.setString(1, arr[0]+" "+arr[1]);
                prest.setString(2, arr[2]);
                prest.setString(3, arr[3]);
                prest.setString(4, arr[4]);
                prest.setString(5, arr[5]);
                prest.setString(6, arr[6]);
                prest.setString(7, arr[7]);
                prest.setString(8, arr[8]);
                prest.setString(9, arr[9]);
                prest.setString(10, arr[10]);
                prest.setString(11, arr[11]);
                prest.setString(12, arr[12]);
                prest.setString(13, arr[13]);
                //添加到批處理
                prest.addBatch();    
            }

            int [] intarr = prest.executeBatch();
            conn.commit();  
            prest.clearBatch();                      
            prest.close();
            conn.close();
            for (int j = 0 ; j < intarr.length; j++) {
                if (intarr[j] > 0) {
                    count +=1;
                }
            }    
        } catch (Exception e) {
            System.out.println(new Date().toLocaleString()+":數據庫插入操作失敗"+e.getMessage());
        }      
        System.out.println("本次操作成功插入"+count+"行數據");    
    }
    
    /**
     * 創建連接池
     * @return Connection
     */
    public static Connection getConnection() {
        Connection con = null;
        try {
            //從配置文件中獲取jdbc config
            JDBCInfo jdbc = JsonFileUtils.readJsonFile(new File("resource/config.json"), JDBCInfo.class);
            if (jdbc != null) {
                //mysql驅動加載
                Class.forName(jdbc.getDriver());
                con = DriverManager.getConnection(jdbc.getUrl(),
                        jdbc.getUser(), jdbc.getPassword());                            
            }
        } catch (Exception e) {
            System.out.println("數據庫連接失敗" +e.getMessage());
        }
        return con;
    }
    
    
    /**
     * 獲取Sql
     * @param arr
     * @return
     */
    public String getSql(String[] arr) {
        
        StringBuffer sql = new StringBuffer("INSERT INTO log_info (");
        sql.append("date_time,");
        sql.append("s_sitename,");
        sql.append("s_ip,");
        sql.append("cs_method,");
        sql.append("cs_uri_stem,");
        sql.append("cs_uri_query,");
        sql.append("s_port,");
        sql.append("cs_username,");
        sql.append("c_ip,");
        sql.append("cs_user_agent,");
        sql.append("sc_status,");
        sql.append("sc_substatus,");
        sql.append("sc_win32_status");
        sql.append(") VALUES ('");
        sql.append(arr[0]+" "+arr[1]);
        sql.append("','");
        sql.append(arr[2]);
        sql.append("','");
        sql.append(arr[3]);
        sql.append("','");
        sql.append(arr[4]);
        sql.append("','");
        sql.append(arr[5]);
        sql.append("','");
        sql.append(arr[6]);
        sql.append("','");
        sql.append(arr[7]);
        sql.append("','");
        sql.append(arr[8]);
        sql.append("','");
        sql.append(arr[9]);
        sql.append("','");
        sql.append(arr[10]);
        sql.append("','");
        sql.append(arr[11]);
        sql.append("','");
        sql.append(arr[12]);
        sql.append("','");
        sql.append(arr[13]);
        sql.append("')");
        
        return sql.toString();
    }
}

　　調用代碼：

    /**
     * 此方法采用遞歸操作，直至數據全部入庫寫入完畢
     * 同時調用5個線程進行入庫操作
     * @param data
     * @param start
     * @param end
     */
    public static void threadsHandle(List<String> data, int start, int end) {
        
        int total = data.size();
        int size = (int)data.size()/5;
        //數據不越界
        if (start < total) {
            List<String> temp = null;
            if (end < total) {
                temp = data.subList(start, end);                        
            } else if (end >= total) {
                temp = data.subList(start, total);
            }
            //執行數據寫入
            DBSource thread = new DBSource(temp);
            thread.start();

            start = end;
            end = start+size;
            threadsHandle(data, start, end);
        }
    }

 

最終結果：

     原來的12分鍾，變成了6秒左右，效率大了一大截。其他朋友如果有更好的建議，可以跟我交流下0.0。下次再把數據弄的更大些。