4

Following is some parts of my code, which uses Threading. The purpose is to retrieve all the records from database (approx. 5,00,000) and send them alert email messages. The problem I am facing is the variable emailRecords becomes very heavy and too much time is taken to send email message. How can I make it fast by using multi-threading such that 5,00,000 records are processed parallelly? I tried to use ExecutorService but got confused in implementing it. I got mixed up in the method checkName(), getRecords() and sendAlert(). All these 3 methods are used relevantly. So, where to use executorService ??

Please provide me the suggestion how to proceed with the following code and which part needs editing? Thanks in advance!! 

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 
        }
    }
}
Improve this question
1

3 Answers 3

Reset to default
1

From what i can tell is that you would most likely be single threaded data retrieval, and multi-threaded for the e-mail sending. Roughly, you'd be cycling through your result set and building a list of records. When that list hits a certain size, you make a copy and send off that copy to be processed in a thread, and clear the original list. At the end of the result set, check to see if you have unprocessed records in your list, and send that to the pool as well.

Finally, wait for the threadpool to finish processing all records.

Something along these lines:

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);
   }


}
Improve this answer
Sign up to request clarification or add additional context in comments.

4 Comments

Thanks a lot. Some questions, when 10000(I limited the sql query) records are retrieved from database at a time, then only 1000 is loaded to emaillist and it is processed. We have mentioned 10 max threads, so will there be 10 threads working on these 1000 records at a time? such that in a single threaded, if 1 record takes 2 seconds to process, 10,000 will take 20,000 seconds. But now after implementing this, 10 records will take 2 second, 10,000 will take 2000 seconds? And after 1000 records are processed, then it loads another 1000 records and proceeds until it reaches to 10,000??
The way it's set up, you could be executing up to 11 blocks of 1000 concurrently. 10 in the thread pool, and one in the main thread (as per the CallerRunsPolicy). You could very well set this up differently (say with a SynchronousQueue instead of ArrayBlockingQueue) to block submission until threads are available, but I don't want to get into a full discussion on this, as there's many great resources you can google for to find this stuff.
PS: In addition to having 11 concurrently executing blocks, you could have 5 blocks sitting in the queue (i gave the ArrayBlockingQueue a capacity of 5). WHen the queue fills up, and you submit another item, the CallerRunsPolicy kicks in. So even with 1 million records, you wuold only ever have 16 blocks loaded at a time.
thanks. Currently I am going to use ArrayBlockingQueue and research about SynchronousQueue. Also, can you please guide me how to check which thread is taking how much time to process?
1

The advantage of using the FixedThreadPool is that you don't have to do the expensive process of creating the threads again and again, its done at the beginning...see below..

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..

  }

 }
Improve this answer

4 Comments

I think the problem OP is attempting to avoid is having to queue up all the records in a List first before doing any work on that list (like sending email)
@Kumar, thanks. But arList can only be filled after validation of Country object and it retrieves email addresses of that particular country. That's why it is inside the method getResult() which is called from checkName() after validation. I am getting stucked as how to implement executorService that includes all of these methods.. or at least sends email parallelly.
@MadanMadan can you do something like validating the Country object and then retrive its email and store it in the ArrayList.... till you have done it all.... then use Executors as i mentioned above to make those emails parallelly send.... Or you can do something like this..... after validating a single country object and retriving its email, sent it to on executor to be executed, while you keep retriving the emails...
@kumar you mean to say inside the method getRecords() I should use ExecutorService when calling the method sendAlert()??
0

Creating a second thread to do all of the work in instead of doing the same work in the main thread isn't going to help you avoid the problem of filling up the emailRecords list with 5 million records before processing any of them.

It sounds like your goal is to be able to read from the database and send email in parallel. Instead of worrying about the code, first think of an algorithm for the work you want to accomplish. Something like this:

  1. In one thread, query for the records from the database, and for each result, add one job to an ExecutorService
  2. That job sends email to one person/address/record.

or alternatively

  1. Read records from the database in batches of N (50, 100, 1000, etc)
  2. Submit each batch to the executorService
Improve this answer

3 Comments

So the code you posted doesn't fit the algorithm at all. You would typically want at least two classes to handle this: the db-querying logic (which has a reference to the ExecutorService) and the subclass of Runnable/Callable which you will instantiate in the first class with all the data it needs to do it's work and which you pass to ExecutorService.submit()
@mattThanks. I can limit query result to 10000 at a time. But I am confused where to use ExecutorService? In the code, 1. the thread's public void run() method calls checkName() method where Country object is initialized. 2. checkName() method calls getRecords() method where database query runs and fills up the emailRecords 3. And if emailRecords>0, sendAlert() method is called. I am stucked here. 10,000 records at a time is retrieved but how can we use executorService for the method sendAlert() so that it is used parallelly without collission? If you can, just an example please..
In my last comment I mentioned how you could structure this. Throw away the old code and concerns about method A calls B calls C calls D. Have a class that handles the DB query have a reference to the executorService and submit new jobs to it while processing the query results.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.