如何实现MySQL多查询无等待并发执行？实时流数据入库需求

我运行一个Socket客户端获取持续的流数据，需要实时写入远程数据库，但当前的查询执行是顺序执行的。假设到远程MySQL服务器的PING延迟为1秒，执行100条查询要耗时100秒。我需要100条查询在1秒内完成，无需等待结果（推送到数据库服务器后就不用管，忽略网络延迟），而且必须实时执行每条查询，批量INSERT不可行，拿到数据就得立刻写入。要求满足：

• 非阻塞、非顺序
• 同时、并发、并行

当前的Node.js代码示例：

function dbQuery(sql)
{
    dbcon.query(sql, function(err, result) { 
        // 假设到DB服务器的PING是1秒
        // 如果逐条执行，100条查询要花100秒
    });
}

for (var i=0; i<100; i++) {
    dbQuery("INSERT INTO table VALUES some values");
}

Node.js完全可以实现需求，利用其异步非阻塞特性+数据库连接池，就能让100条查询几乎同时发送到数据库，总耗时接近单次PING延迟。

关键调整点 #

1. 使用带连接池的高效MySQL驱动（比如mysql2），连接池允许同时发起多个TCP连接，突破单连接的顺序执行限制。
2. 不等待查询回调结果，直接触发请求，让所有查询在极短时间内被发送出去。

修改后的示例代码：

const mysql = require('mysql2/promise');

// 创建连接池，设置足够的并发连接数
const pool = mysql.createPool({
  host: '你的数据库地址',
  user: '用户名',
  password: '密码',
  database: '目标库',
  waitForConnections: false, // 连接池满时直接抛出错误，避免等待
  connectionLimit: 100,
  queueLimit: 0
});

async function dbQuery(sql) {
  // 直接从连接池拿连接执行查询，忽略返回结果
  pool.query(sql).catch(err => {
    // 可选：记录错误，不中断流程
    console.error('查询出错:', err);
  });
}

// 循环发起100条并发查询
for (let i = 0; i < 100; i++) {
  dbQuery(`INSERT INTO table VALUES (${i}, 'value-${i}')`);
}

如果需要多线程/协程模型的方案，以下语言也能轻松满足需求：

Java #

用线程池+JDBC连接池（比如HikariCP）实现并行提交，每个查询由线程池中的独立线程处理：

import com.zaxxer.hikari.HikariDataSource;
import java.sql.Connection;
import java.sql.Statement;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DbConcurrentInsert {
    private static final HikariDataSource dataSource;

    static {
        dataSource = new HikariDataSource();
        dataSource.setJdbcUrl("jdbc:mysql://你的数据库地址:3306/目标库");
        dataSource.setUsername("用户名");
        dataSource.setPassword("密码");
        dataSource.setMaximumPoolSize(100);
    }

    public static void dbQuery(String sql) {
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement()) {
            stmt.executeUpdate(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(100);
        for (int i = 0; i < 100; i++) {
            final int index = i;
            executor.submit(() -> dbQuery("INSERT INTO table VALUES (" + index + ", 'value-" + index + "')"));
        }
        executor.shutdown();
    }
}

Python #

用**ThreadPoolExecutor+数据库连接池**（比如DBUtils），避开GIL对IO操作的限制：

from concurrent.futures import ThreadPoolExecutor
from DBUtils.PooledDB import PooledDB
import pymysql

# 创建连接池
pool = PooledDB(
    creator=pymysql,
    host='你的数据库地址',
    user='用户名',
    password='密码',
    database='目标库',
    maxconnections=100
)

def db_query(sql):
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            conn.commit()
    except Exception as e:
        print(f"查询出错: {e}")
        conn.rollback()
    finally:
        conn.close()

if __name__ == '__main__':
    with ThreadPoolExecutor(max_workers=100) as executor:
        for i in range(100):
            executor.submit(db_query, f"INSERT INTO table VALUES ({i}, 'value-{i}')")

PHP #

用Swoole协程实现轻量并发，协程比线程资源消耗更低：

<?php
use Swoole\Coroutine;
use Swoole\Coroutine\MySQL;

Coroutine\run(function () {
    $config = [
        'host' => '你的数据库地址',
        'user' => '用户名',
        'password' => '密码',
        'database' => '目标库',
    ];

    for ($i = 0; $i < 100; $i++) {
        Coroutine::create(function ($i) use ($config) {
            $mysql = new MySQL();
            $mysql->connect($config);
            $sql = "INSERT INTO table VALUES ($i, 'value-$i')";
            $mysql->query($sql);
            $mysql->close();
        }, $i);
    }
});
?>

• 需调整MySQL服务器的max_connections参数，确保能支撑至少100个并发连接（默认值可能不足）。
• 即使不等待结果，也要监控数据库写入性能，避免短时间请求过载导致服务异常。
• 建议记录错误日志，方便后续排查问题，但不要中断并发流程。

内容的提问来源于stack exchange，提问作者neobie