出自:

騰訊課堂 700多分鐘干貨實戰Java多執行緒高并發高性能實戰全集 , 我學習完了之后,我做了個筆記.

背景

某應用程式(單臺服務器,非分布式的多臺服務器),這單臺服務器就是你的筆記本電腦了,
?

并發產生100萬條資料,這100w條資料是你自己產生的,假設你是架構師,如何運用多執行緒等基礎知識將這100萬條資料,快速同步(4分鐘以內)到MySQL資料庫?
?

分析百萬資料快速入庫的特點

1.百萬資料快速入庫的特點:
資料量比較大(高并發),時間很短(性能),
100萬條資料如果一條一條的插入到資料庫的話,時間是很慢的,所以我們采用批量的方式插入,每次分一兩萬, 分多個批次,并行的插入到資料庫里面.
這就是用并發編程的方式去解決高并發高性能的問題
2.百萬資料如何在短時間內入庫?如何從架構角度優化性能?
應用程式怎么優化呢? 可以采用并發編程的形式,比如說多執行緒,執行緒池去提升性能
在資料連接池這層,我們可以調優,讓它的并發量更高,提高資料庫連接池的整體性能.
?

代碼

Producer

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class Producer {

    public static void main(String[] args) {
        Producer.createData();
    }

    public static void createData() {
        ExecutorService pool = Executors.newFixedThreadPool(100);
        final int totalPageNo = 50; //分50批次

        final int pageSize = 20000; //每頁大小是2萬條
        //共10w條資料，每頁5000條資料，20個執行緒
        final long start = System.currentTimeMillis();
        final AtomicInteger atomicInt = new AtomicInteger();
        for (int currentPageNo = 0; currentPageNo < totalPageNo; currentPageNo++) {
            final int finalCurrentPageNo = currentPageNo;

            Runnable run = new Runnable() {

                @Override
                public void run() {
                    List userList = new ArrayList<>();
                    for (int i = 1; i <= pageSize; i++) {
                        int id = i + finalCurrentPageNo * pageSize;
                        User user = new User();
                        user.setId(id);
                        user.setName("huanglaoxie:" + id);
                        userList.add(user);
                    }

                    atomicInt.addAndGet(UserBatchHandler.batchSave(userList, Thread.currentThread().getName()));
                    //入庫的資料達到一百萬條的時候就會有個統計.
                    if (atomicInt.get() == (totalPageNo * pageSize)) {
                        //如果有一百萬的時候.就會在這里有個結果
                        System.out.println("同步資料到db，它已經花費 " + ((System.currentTimeMillis() - start) / 1000) + "  秒");
                    }

                }
            };
            try {
                Thread.sleep(5);
            } catch (InterruptedException e) {

                e.printStackTrace();
            }
            pool.execute(run);
        }

    }

}

User

import java.sql.Timestamp;

public class User {
    private int id;
    private String name;
    private Timestamp createdTime;
    private Timestamp updatedTime;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Timestamp getCreatedTime() {
        return createdTime;
    }

    public void setCreatedTime(Timestamp createdTime) {
        this.createdTime = createdTime;
    }

    public Timestamp getUpdatedTime() {
        return updatedTime;
    }

    public void setUpdatedTime(Timestamp updatedTime) {
        this.updatedTime = updatedTime;
    }
}

UserBatchHandler

 

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

public class UserBatchHandler {


    public static int batchSave(List userList, String threadName)  {
        String insertSql ="INSERT INTO user(id,name,createdTime,updatedTime) VALUES(?,?,sysdate(),sysdate())";
        //取得發送sql陳述句的物件
        PreparedStatement pst = null;
        User  user;
        int[] count = new int[0];
        Connection conn = null;
        try {
            conn= DataSourceUtils.getConnection();
            pst = conn.prepareStatement(insertSql);

            long start=System.currentTimeMillis();
            if(null!=userList&&userList.size()>0){
                for(int i=0;i<userList.size();i++){
                    user= (User) userList.get(i);
                    pst.setInt(1,user.getId());
                    pst.setString(2,user.getName());
                    //加入批處理
                    pst.addBatch();
                }

                count= pst.executeBatch();
                System.out.println(count.length);
                System.out.println(" threadName為"+threadName+", sync data to db, it  has spent " +(System.currentTimeMillis()-start)+"  ms");
        }
        } catch (SQLException e) {
            e.printStackTrace();
        }finally {
            //4. 釋放資源
            DataSourceUtils.close(conn, pst);
        }

        //獲取到資料更新的行數
        return count.length;
    }
}

DataSourceUtils

 

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;

public class DataSourceUtils {

    public static void main(String[] args){
      Connection conn=  DataSourceUtils.getConnection();
      System.out.println("conn is  :  "+conn);
    }

    //創建一個成員變數
    private static DataSource ds;

    /**
     * 加載的代碼寫在靜態代碼塊中
     */
    static {
        try {
            Properties info = new Properties();
            //加載類路徑下，即src目錄下的druid.properties這個檔案
            info.load(DataSourceUtils.class.getResourceAsStream("/druid.properties"));

            //讀取屬性檔案創建連接池
            ds = DruidDataSourceFactory.createDataSource(info);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 得到資料源
     */
    public static DataSource getDataSource() {
        return ds;
    }

    /**
     * 得到連接物件
     */
    public static Connection getConnection() {
        try {
            return ds.getConnection();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 釋放資源
     */
    public static void close(Connection conn, Statement stmt, ResultSet rs) {
        if (rs!=null) {
            try {
                rs.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (stmt!=null) {
            try {
                stmt.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (conn!=null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }


    public static void close(Connection conn, Statement stmt) {
        close(conn, stmt, null);
    }


}

druid.properties

# 配置連接池的引數
initialSize=50
maxActive=200
maxWait=600000
minIdle=5



driverClassName=com.mysql.jdbc.Driver
url=jdbc:mysql://zjj101:3306/test?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC&useSSL=false
username=root
password=root

sql腳本

CREATE TABLE `test`.`user` (
  `id` INT NOT NULL,
  `name` VARCHAR(45) NULL,
  `createdTime` timestamp NULL,
  `updatedTime` timestamp NULL,
  PRIMARY KEY (`id`))
COMMENT = '用戶測驗表';

ALTER TABLE `test`.`user`
ADD INDEX `index` (`id` ASC);


SELECT count(*) FROM test.user;
# delete  from test.user;
SELECT *  FROM test.user  order by  id desc;

操作說明:

執行sql腳本
2.執行Producer類即可
?

其它優化:

池技術為什么能提升性能?

連接池:
tomcat連接池,資料庫連接池等等,通過復用連接來減少創建和釋放連接的時間來提升性能.
執行緒池:
執行緒池和連接池也是一樣的,通過復用連接來減少創建和釋放連接的時間來提升性能.
?

druid資料庫連接池性能調優

 # 配置連接池的引數
 initialSize=50
 # 連接池的最大資料庫連接數，設為0表示無限制，
 maxActive=200
 # 最大建立連接等待時間，如果超過此時間將接到例外，設為-1表示無限制，
 maxWait=600000
 # 連接池中的最小空閑連接數，Druid會定時掃描連接池的連接，如果空閑的連接數大于該值，則關閉多余的連接，反之則創建更多的連接以滿足最小連接數要求，
 minIdle=5

MySQL的核心引數優化

配置 “my.cnf” 檔案里面的innodb_thread_concurrency 的配置,這個是調整執行緒的并發數 ,配置完了別忘了重啟MySQL服務
?

當 innodb_thread_concurrency=12
?

執行程式結果:
我測驗一下,第一次是 79秒第二次88秒第三次92秒, 不知道為什么一次比一次多了.
?

當innodb_thread_concurrency=32的時候
我測驗了一下第一次是67秒 ,第二次是62秒
?

另外,其它引數也可以修改:
innodb_buffer_pool_size 引數
max_allowed_packet 引數配置

代碼Git地址

https://gitee.com/zjj19941/mutil-thread.git

看case1

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/312498.html

標籤：其他

上一篇：如何通過介面訪問GameObject

下一篇：四元數y軸90度旋轉的基本概念

Java多執行緒百萬資料快速入庫實戰

出自:

背景