Java中如何用线程Vector实现数据库数据的并发按需处理？

核心设计思路 #

• 自定义DataProcessingThread：封装数据处理逻辑，维护自身工作状态（是否正在处理），提供数据分配与启动处理的方法
• 主线程职责：初始化容量为5的工作线程池，循环执行以下逻辑：
  1. 统计当前空闲线程数量，从数据库拉取对应数量的数据
  2. 将数据分配给空闲线程，触发处理
  3. 休眠10秒后重复上述步骤，直到所有数据库数据处理完成

完整代码实现 #

1. 自定义工作线程类 #

public class DataProcessingThread extends Thread {
    // 线程安全的状态标记：false=空闲，true=正在处理
    private volatile boolean isProcessing;
    private String dataName;
    private String dataItem;

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            synchronized (this) {
                // 等待主线程分配数据
                while (!isProcessing) {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                }
            }

            // 替换为实际业务处理逻辑
            try {
                System.out.println("线程" + Thread.currentThread().getName() + "处理数据：" + dataName + "-" + dataItem);
                Thread.sleep(2000); // 模拟处理耗时
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } finally {
                // 处理完成，标记为空闲
                synchronized (this) {
                    isProcessing = false;
                    dataName = null;
                    dataItem = null;
                }
            }
        }
    }

    // 分配数据并唤醒线程处理
    public synchronized void assignData(String dataName, String dataItem) {
        this.dataName = dataName;
        this.dataItem = dataItem;
        this.isProcessing = true;
        notify();
    }

    // 线程安全的状态检查
    public synchronized boolean isProcessing() {
        return isProcessing;
    }
}

2. 主线程调度逻辑 #

import java.sql.*;
import java.util.Vector;

public class MainThreadScheduler {
    // 替换为实际数据库配置
    private static final String DB_URL = "jdbc:mysql://localhost:3306/your_db";
    private static final String DB_USER = "root";
    private static final String DB_PASSWORD = "password";

    public static void main(String[] args) {
        // 初始化5个工作线程
        Vector<DataProcessingThread> threadVector = new Vector<>(5);
        for (int i = 0; i < 5; i++) {
            DataProcessingThread thread = new DataProcessingThread();
            thread.setName("Worker-" + (i+1));
            thread.start();
            threadVector.add(thread);
        }

        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet rs = null;

        try {
            // 初始化数据库连接
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD);
            String selectQuery = "SELECT dataName, dataItem FROM thread";
            preparedStatement = connection.prepareStatement(selectQuery);
            rs = preparedStatement.executeQuery();

            boolean hasMoreData = true;
            while (hasMoreData) {
                // 统计空闲线程数
                int idleCount = 0;
                for (DataProcessingThread thread : threadVector) {
                    if (!thread.isProcessing()) {
                        idleCount++;
                    }
                }

                if (idleCount == 0) {
                    // 无空闲线程，短暂等待后重试
                    Thread.sleep(500);
                    continue;
                }

                // 为空闲线程分配数据
                int assignedCount = 0;
                while (assignedCount < idleCount && rs.next()) {
                    String dataName = rs.getString("dataName");
                    String dataItem = rs.getString("dataItem");
                    // 找到第一个空闲线程分配数据
                    for (DataProcessingThread thread : threadVector) {
                        if (!thread.isProcessing()) {
                            thread.assignData(dataName, dataItem);
                            assignedCount++;
                            break;
                        }
                    }
                }

                // 检查是否还有未处理数据
                hasMoreData = rs.next();
                if (hasMoreData) {
                    rs.previous(); // 回退ResultSet指针，避免跳过数据
                }

                // 休眠10秒后再次检查
                Thread.sleep(10000);
            }

        } catch (ClassNotFoundException | SQLException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 释放数据库资源
            try {
                if (rs != null) rs.close();
                if (preparedStatement != null) preparedStatement.close();
                if (connection != null) connection.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }

            // 中断所有工作线程，避免线程泄漏
            for (DataProcessingThread thread : threadVector) {
                thread.interrupt();
            }
        }
    }
}

关键注意事项 #

• 线程安全：isProcessing用volatile修饰，结合synchronized块确保状态检查与修改的原子性，避免多线程下的状态不一致
• 资源管理：必须在finally块中关闭数据库连接、Statement和ResultSet，防止资源泄漏
• 线程中断：工作线程响应中断信号，主线程结束时主动中断所有工作线程，避免线程长期存活
• ResultSet处理：通过rs.next()和rs.previous()控制数据读取位置，实现按需拉取分配

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