Java多线程MySQL批量插入方案遇问题，寻求技术解决方法

嘿，针对你遇到的多线程MySQL批量插入104错误和内存优化问题，我来分享一些实战经验——毕竟我之前也踩过类似的坑😉

首先明确：MySQL错误码104最常见的场景是连接被重置（Connection reset by peer）或者连接超时断开，结合你的多线程批量插入逻辑，大概率是这几个原因导致的：

一、揪出104错误的核心诱因 #

• 连接资源线程不安全：如果你的MYSQLInsertThread和数据收集线程共享了同一个MySQL连接句柄，麻烦就大了——MySQL的连接本身不是线程安全的，多线程同时操作会导致连接状态混乱，服务器端直接重置连接是常有的事。
• 连接超时被踢：如果数据收集的间隔太长，插入线程的MySQL连接可能因为长时间闲置，触发了服务器端的wait_timeout或interactive_timeout参数，连接被自动断开，这时候再用这个连接插入就会抛出104错误。
• 批量数据包超限：当你把batchCount设为100时，生成的批量插入SQL可能超过了MySQL的max_allowed_packet限制，服务器接收到超大数据包直接拒绝，也会触发连接重置。

二、针对性解决步骤 #

1. 给每个线程分配独立的连接 #

绝对不要共享连接！修改你的MYSQLInsertThread实现：

• 让插入线程在启动时自己创建新的MySQL连接，任务结束后（或者空闲超时后）正确关闭；
• 如果用连接池，确保每个线程从池里拿独立的连接，用完及时归还。

给你个伪代码参考（以Python为例，可适配自己的开发语言）：

class MYSQLInsertThread(threading.Thread):
    def __init__(self, batch_queue):
        super().__init__()
        self.batch_queue = batch_queue
        self.conn = None
        self.cursor = None
        self.db_config = {"host": "xxx", "user": "xxx", "password": "xxx", "database": "xxx"}

    def run(self):
        # 线程启动时初始化独立连接
        self._init_conn()
        while True:
            batch_data = self.batch_queue.get()
            if batch_data is None:  # 收到终止信号就退出
                break
            try:
                # 执行批量插入
                insert_sql = "INSERT INTO your_table (col1, col2) VALUES (%s, %s)"
                self.cursor.executemany(insert_sql, batch_data)
                self.conn.commit()
            except Exception as e:
                if "104" in str(e):
                    # 连接断了，重新连了再试
                    self._reconnect()
                    self.cursor.executemany(insert_sql, batch_data)
                    self.conn.commit()
                else:
                    # 其他错误按需处理
                    print(f"插入失败: {e}")
                    self.conn.rollback()
            finally:
                self.batch_queue.task_done()
        # 线程结束前清理连接
        self._close_conn()

    def _init_conn(self):
        self.conn = mysql.connector.connect(**self.db_config)
        self.cursor = self.conn.cursor()

    def _reconnect(self):
        self._close_conn()
        self._init_conn()

    def _close_conn(self):
        if self.cursor:
            self.cursor.close()
        if self.conn:
            self.conn.close()

2. 搞定连接超时问题 #

• 如果你有权限修改MySQL配置，把wait_timeout和interactive_timeout调大一点，比如设为3600秒（1小时），避免短时间内闲置被踢；
• 在插入线程里加个心跳机制，比如每隔10分钟执行一次SELECT 1，保持连接活跃；
• 每次执行插入前先检查连接状态，比如执行个简单的查询，失败就重新连接。

3. 调整数据包大小限制 #

• 先查一下当前的限制：SHOW VARIABLES LIKE 'max_allowed_packet';
• 如果批量SQL超过了这个值，临时调整的话可以执行SET GLOBAL max_allowed_packet=67108864;（也就是64MB），要永久生效的话修改my.cnf/my.ini配置文件，添加max_allowed_packet=64M。

三、数百万条插入的内存优化技巧 #

要控制内存占用，这几个点一定要注意：

• 用阻塞队列做缓冲：数据收集线程把数据放到有容量限制的阻塞队列里，队列满了收集线程就会暂停，避免无限制堆积数据把内存撑爆；
• 合理设置批次大小：不是batchCount越大越好，太大的批次会让单条SQL过长（容易触发数据包限制），还会占用更多内存存批次数据。建议根据单条数据大小调整，比如每条1KB的话，1000条就是1MB，这个量级比较均衡；
• 只存必要数据：收集数据的时候别带无关字段，只保留插入需要的内容，减少每条数据的内存占用；
• 流式处理：如果数据是流式来源（比如日志、网络流），别把所有数据都加载到内存，边收集边批量插入；
• 及时释放内存：批量插入完成后，立刻清空存储批次数据的容器（比如列表、数组），让垃圾回收机制及时回收内存。

最后验证小建议 #

1. 先单独测试插入线程的批量插入功能，确保单线程下batchCount=100能正常跑；
2. 再开启数据收集线程，观察队列的内存占用和连接状态；
3. 去看MySQL的错误日志（一般在/var/log/mysql/error.log或者对应的路径），里面会有104错误的详细触发原因，能帮你精准定位问题。

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