Java，多线程，ExecutorService

Question

以下是我的代码的某些部分，它使用了线程。目的是从数据库中检索所有记录（大约 5,00,000 条）并向它们发送警报电子邮件消息。我面临的问题是变量 emailRecords 变得非常繁重，并且花费了太多时间来发送电子邮件。如何通过使用多线程来快速处理，以便并行处理 5,00,000 条记录？我尝试使用 ExecutorService 但在实现时感到困惑。我在方法 checkName()、getRecords() 和 sendAlert() 中搞混了。所有这三种方法都被相关地使用。那么，executorService在哪里使用呢？？

请向我提供如何处理以下代码以及需要编辑哪个部分的建议？提前致谢！！

public class sampledaemon implements Runnable {

    private static List<String[]> emailRecords = new ArrayList<String[]>();

    public static void main(String[] args) {
        if (args.length != 1) {
            return;
        }

        countryName = args[0];

        try {
            Thread t = null;
            sampledaemon daemon = new sampledaemon();
            t = new Thread(daemon);
            t.start();
        } catch (Exception e) {
            e.printStackTrace()
        }

    }

    public void run() {
        Thread thisThread = Thread.currentThread();
        try {
            while (true) {
                checkName(countryName);
                Thread.sleep(TimeUnit.SECONDS.toMillis(10));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void checkName(String countryName) throws Exception {
        Country country = CountryPojo.getDetails(countryName)

        if (country != null) {
            getRecords(countryconnection);
        }
    }

    private void getRecords(Country country, Connection con) {
        String users[] = null;
        while (rs.next()) {
            users = new String[2];
            users[0] = rs.getString("userid");
            users[1] = rs.getString("emailAddress");
            emailRecords.add(props);

            if (emailRecords.size() > 0) {
                sendAlert(date, con);
            }
        }
    }

    void sendAlert(String date, Connection con) {
        for (int k = 0; k < emailRecords.size(); k++) {
            //check the emailRecords and send email 
        }
    }
}

Answer 1

据我所知，您很可能是单线程数据检索，而多线程用于电子邮件发送。粗略地说，您将循环浏览您的结果集并建立一个记录列表。当该列表达到一定大小时，您制作一个副本并将该副本发送到线程中进行处理，并清除原始列表。在结果集的末尾，检查列表中是否有未处理的记录，并将其发送到池中。

最后，等待线程池处理完所有记录。

类似的东西：

protected void processRecords(String countryName) {
  ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 10, TimeUnit.SECONDS, 
        new ArrayBlockingQueue<Runnable>(5), new ThreadPoolExecutor.CallerRunsPolicy());

   List<String[]> emaillist = new ArrayList<String>(1000);

   ResultSet rs = ....

   try {
     while (rs.next()) {
        String user[] = new String[2];
        users[0] = rs.getString("userid");
        users[1] = rs.getString("emailAddress");

        emaillist.add(user);
        if (emaillist.size() == 1000) {
            final List<String[]> elist = new ArrayList<String[]>(emaillist);
            executor.execute(new Runnable() {
                public void run() {
                    sendMail(elist);
                }
            }
            emaillist.clear();
        }
     }
   }
   finally {
     DbUtils.close(rs);
   }

   if (! emaillist.isEmpty()) {
            final List<String[]> elist = emaillist;
            executor.execute(new Runnable() {
                public void run() {
                    sendMail(elist);
                }
            }
            emaillist.clear();
   }

   // wait for all the e-mails to finish.
   while (! executor.isTerminated()) {
       executor.shutdown();
       executor.awaitTermination(10, TimeUnit.DAYS);
   }


}

Answer 2

使用FixedThreadPool 的好处是您不必一次又一次地进行昂贵的线程创建过程，它一开始就完成 ...见下文..

ExecutorService executor = Executors.newFixedThreadPool(100);

ArrayList<String> arList =  Here your Email addresses from DB will go in ;

for(String s : arList){

     executor.execute(new EmailAlert(s));

 }



public class EmailAlert implements Runnable{

  String addr;

   public EmailAlert(String eAddr){


         this.addr = eAddr;

       }


  public void run(){


     // Do the process of sending the email here..

  }

 }

Answer 3

创建第二个线程来完成所有工作而不是在主线程中完成相同的工作并不能帮助您避免在处理任何一个之前用 500 万条记录填充 emailRecords 列表的问题他们。

听起来您的目标是能够从数据库中读取数据并并行发送电子邮件。与其担心代码，不如先为您想要完成的工作考虑一种算法。像这样的：

1. 在一个线程中，从数据库中查询记录，为每个结果，添加一个作业到 ExecutorService
2. 该工作向一个人/地址/记录发送电子邮件。

或者

1. 分批从数据库中读取记录 N（50、100、1000 等）
2. 将每个批次提交给 executorService