多線程資料庫

⑴ 要瘋了，怎樣用多線程向MYsql資料庫中寫入數據

在MySQL 8.0 之前， 我們假設一下有一條爛SQL，

mysqlselect * from t1 order by rand() ;

以多個線程在跑，導致CPU被跑滿了，其他的請求只能被阻塞進不來。那這種情況怎麼辦？

大概有以下幾種解決辦法：

• 設置max_execution_time 來阻止太長的讀SQL。那可能存在的問題是會把所有長SQL都給KILL 掉。有些必須要執行很長時間的也會被誤殺。

• 自己寫個腳本檢測這類語句，比如order by rand()， 超過一定時間用Kill query thread_id 給殺掉。

那能不能不要殺掉而讓他正常運行，但是又不影響其他的請求呢？

那mysql 8.0 引入的資源組（resource group，後面簡寫微RG）可以基本上解決這類問題。

比如我可以用 RG 來在SQL層面給他限制在特定的一個CPU核上，這樣我就不管他，讓他繼續運行，如果有新的此類語句，讓他排隊好了。

為什麼說基本呢？目前只能綁定CPU資源，其他的暫時不行。

那我來演示下如何使用RG。

創建一個資源組user_ytt. 這里解釋下各個參數的含義，

• type = user 表示這是一個用戶態線程，也就是前台的請求線程。如果type=system，表示後台線程，用來限制mysql自己的線程，比如Innodb purge thread,innodb read thread等等。

• vcpu 代表cpu的邏輯核數，這里0-1代表前兩個核被綁定到這個RG。可以用lscpu，top等列出自己的CPU相關信息。

• thread_priority 設置優先順序。user 級優先順序設置大於0。

• mysqlmysql> create resource group user_ytt type = user vcpu = 0-1 thread_priority=19 enable;Query OK, 0 rows affected (0.03 sec)




RG相關信息可以從 information_schema.resource_groups 系統表裡檢索。

• mysqlmysql> select * from information_schema.resource_groups;+---------------------+---------------------+------------------------+----------+-----------------+| RESOURCE_GROUP_NAME | RESOURCE_GROUP_TYPE | RESOURCE_GROUP_ENABLED | VCPU_IDS | THREAD_PRIORITY |+---------------------+---------------------+------------------------+----------+-----------------+| USR_default | USER | 1 | 0-3 | 0 || SYS_default | SYSTEM | 1 | 0-3 | 0 || user_ytt | USER | 1 | 0-1 | 19 |+---------------------+---------------------+------------------------+----------+-----------------+3 rows in set (0.00 sec)




我們來給語句select guid from t1 group by left(guid,8) order by rand() 賦予RG user_ytt。

• mysql> show processlist;+-----+-----------------+-----------+------+---------+-------+------------------------+-----------------------------------------------------------+| Id | User | Host | db | Command | Time | State | Info |+-----+-----------------+-----------+------+---------+-------+------------------------+-----------------------------------------------------------+| 4 | event_scheler | localhost | NULL | Daemon | 10179 | Waiting on empty queue | NULL || 240 | root | localhost | ytt | Query | 101 | Creating sort index | select guid from t1 group by left(guid,8) order by rand() || 245 | root | localhost | ytt | Query | 0 | starting | show processlist |+-----+-----------------+-----------+------+---------+-------+------------------------+-----------------------------------------------------------+3 rows in set (0.00 sec)




找到連接240對應的thread_id。

• mysqlmysql> select thread_id from performance_schema.threads where processlist_id = 240;+-----------+| thread_id |+-----------+| 278 |+-----------+1 row in set (0.00 sec)




給這個線程278賦予RG user_ytt。沒報錯就算成功了。

• mysqlmysql> set resource group user_ytt for 278;Query OK, 0 rows affected (0.00 sec)




當然這個是在運維層面來做的，我們也可以在開發層面結合 MYSQL HINT 來單獨給這個語句賦予RG。比如：

• mysqlmysql> select /*+ resource_group(user_ytt) */guid from t1 group by left(guid,8) order by rand()....8388602 rows in set (4 min 46.09 sec)




RG的限制：

• Linux 平台上需要開啟 CAPSYSNICE 特性。比如我機器上用systemd 給mysql 服務加上

  systemctl edit mysql@80 [Service]AmbientCapabilities=CAP_SYS_NICE

• mysql 線程池開啟後RG失效。

• freebsd,solaris 平台thread_priority 失效。

• 目前只能綁定CPU，不能綁定其他資源。

⑵ MYSQL資料庫如何多線程

1。通過線程的互斥來同步操作資料庫
2。資料庫採用事務處理表中的數據
3。採用共享方式打開資料庫，不是以獨占方式打開資料庫
建立一個mysql連接表加上一個
臨界區
，表結點是這樣的（mysqlcon,bool)，根據實際情況定大小。我用的是10個連接。
當要進行mysql操作時，就從表中取出一個閑置的mysql連接，並把bool量改為true，使用完後改成false,臨界區的做用是保障一個mysql連接一次只能被一個線程使用。

⑶ 請教一個多線程同時查詢資料庫問題

多線程操作，請確保每個線程操作的SQL語句中的表是相對獨立的。 不然，你需要安排線程間的順序，也就是lock代碼段。 同一時間，兩個線程一起跑同一句SQL，而且還操作同一張表，那麼，肯定就會有問題了。
一般這種是因為超出資料庫最大鏈接上限。再建立鏈接，不管緩存多少，會自動隊列消息等待。Timeout時間內沒有鏈接取消無法獲得鏈接許可權。可以將自己的資料庫鏈接個數設置大一些。

⑷ 多線程並發訪問資料庫中不同記錄時應該採用什麼辦法

多個線程在訪問同一個資料庫中的記錄的時候，在大並發的情況下，一定要做好事務機制。要不然後果還是很扎心的。可能遇到這樣的問題的時候首先想到的就是同步鎖機制了。

並發產生的情況

其實在實際項目開發匯總，首先要做的就是避免多個線程共用一個資料庫連接，這樣會很容易出問題，最好是一個線程一個連接。在必要的時候需要線程同步或存儲過程加鎖。如果有比較復雜的事務操作的話就需要加鎖了，如果不加的話就會出現你說的臟讀、死鎖等問題。

⑸ 請教如何進行多線程連接資料庫並寫入數據

#include <QCoreApplication>
#include "thread.h"
#include <QVector>
#include <QDebug>

int main(int argc, char *argv[])
{
QCoreApplication a(argc, argv);

QVector<Thread*> vector;
Thread *thread;

//創建多個線程，並start
for(int i=0;i<10;i++){
thread=new Thread;
vector.append(thread);
thread->set(i);
thread->start();
}

//等待所有線程執行完，然後刪除線程
foreach(thread,vector){
thread->wait();
}
foreach(thread,vector){
delete thread;
}

return a.exec();
}

-------------------------------------------------------------------------------------------------

#include "thread.h"

Thread::Thread(QObject *parent) : QThread(parent)
{
}

void Thread::run()
{
begin();
}

//為每個線程創建一個連接名
void Thread::set(int a)
{
connectionName=QString::number(a);
}

void Thread::connectionDatabase(QString dbName)
{
QSqlDatabasedb=QSqlDatabase::addDatabase("QMYSQL",connectionName);
db.setHostName("localhost");
db.setDatabaseName(dbName);
db.setUserName("root");
db.setPassword("");

if(!db.open())
qDebug()<<"db open fail";
}

void Thread::begin()
{
QString dbName="learnsql";
connectionDatabase(dbName);

QSqlDatabase db=QSqlDatabase::database(connectionName);
db.transaction(); //開啟事物

QSqlQuery query(db);

//向表student中插入10000條數據
for(int i=1;i<=10000;i++){
query.exec("insert into student values(1)");
}

db.commit(); //提交事物
}

⑹ 多線程 連接資料庫,C#多線程寫資料庫

多線程連接資料庫的連接池類：

public static class ConnectionPool
{
private static object locker = new object();
private static Dictionary<string, SqlConnection> Connections = null;
public static SqlConnection GetConnection<T>() where T : class, new()
{
string databaseName = NA.Common.Extensions.GetDatabaseName<T>();
if (string.IsNullOrEmpty(databaseName))
return null;
if (Connections == null)
{
lock (locker)
{
Connections = new Dictionary<string, SqlConnection>();
}
}
string connKey = FindFreeSqlConnection(databaseName);
if (connKey != null)
return Connections[connKey];
else
{
string strconn = NA.Common.Extensions.GetConnectionString<T>();
int poolSize = NA.Common.Extensions.GetConnectionPoolSize<T>();
lock (locker)
{
for (int i = 0; i < poolSize; ++i)
{
SqlConnection conn = new SqlConnection(strconn);
conn.Open();
Connections.Add(databaseName + "_" + i.ToString(), conn);
conn.Close();
}
}
return Connections[FindFreeSqlConnection(databaseName)];
}
}

private static string FindFreeSqlConnection(string databaseName)
{
IEnumerable<string> connKeys = Connections.Keys.Where(item => item.StartsWith(databaseName));
if (connKeys != null && connKeys.Count() > 0)
{
foreach (string key in connKeys)
{
if (Connections[key].State == ConnectionState.Closed)
return key;
}
}
return null;
}
}

附加上其中用到的三個方法：

internal static int GetConnectionPoolSize<T>() where T : class, new()
{
string database = GetDatabaseName<T>();
string[] poolSizeArray = ConfigurationManager.AppSettings["ConnectionsPoolSize"].Split('|');
if (poolSizeArray != null)
{
foreach (string sizeItem in poolSizeArray)
{
string[] sizeItemArray = sizeItem.Split(':');
if (database == sizeItemArray[0])
return int.Parse(sizeItemArray[1]);
}
}
return 50;
}
public static string GetConnectionString<T>() where T : class, new()
{
string tableName = GetTableName<T>();
string[] databaseArray = ConfigurationManager.AppSettings["DatabaseArray"].Split('|');
if (databaseArray != null)
{
foreach (string database in databaseArray)
{
string tableNameList = ConfigurationManager.AppSettings[database];
string[] tables = ConfigurationManager.AppSettings[database].Split('|');
if (tables != null && tables.Length > 0)
if (tables.Contains(tableName))
return ConfigurationManager.ConnectionStrings[database].ConnectionString;
}
}
return string.Empty;
}
public static string GetDatabaseName<T>() where T : class, new()
{
string tableName = GetTableName<T>();
string[] databaseArray = ConfigurationManager.AppSettings["DatabaseArray"].Split('|');
if (databaseArray != null)
{
foreach (string database in databaseArray)
{
string tableNameList = ConfigurationManager.AppSettings[database];
string[] tables = ConfigurationManager.AppSettings[database].Split('|');
if (tables != null && tables.Length > 0)
if (tables.Contains(tableName))
return database;
}
}
return string.Empty;
}