How to add elements to ConcurrentHashMap using ExecutorService

Question

I have a requirement of reading User Information from 2 different sources (db) per userId and storing consolidated information in a Map with key as userId. Users in numbers can vary based on period they have opted for. Group of users may belong to different Period of Year.eg daily, weekly, monthly users.

I used HashMap and LinkedHashMap to get this done. As it slows down the process and to make it faster, I thought of using Threading here.

Reading some tutorials and examples now I am using ConcurrentHashMap and ExecutorService.

1. In cases based on some validation I want to skip the current iteration and move to next User info. It doesnot allow to use continue keyword to use within for loop. Is there any way to achieve same differently within Multithreaded code.

2. Moreover below code piece though it works, but its not significantly that faster than the code without threading which creates doubt if Executor Service is implemented correctly.

3. How do we debug in case we get any error in Multithreaded code. Execution holds at debug point but its not consistent and it does not move to next line with F6.

Can someone point out if I am missing something in the code. Or any other example of simillar use case also can be of great help.

public void getMap() throws UserException
{
    long startTime = System.currentTimeMillis();
    Map<String, Map<Integer, User>> map = new ConcurrentHashMap<String, Map<Integer, User>>();
            //final String key = "";
            try
            {
                final Date todayDate = new Date();
                List<String> applyPeriod = db.getPeriods(todayDate);            
                for (String period : applyPeriod)
                {
                    try 
                    {   
                            final String key = period;
                            List<UserTable1> eligibleUsers = db.findAllUsers(key);
                            Map<Integer, User> userIdMap = new ConcurrentHashMap<Integer, User>();
                            ExecutorService executor = Executors.newFixedThreadPool(eligibleUsers.size());
                            CompletionService<User> cs = new ExecutorCompletionService<User>(executor);

                            int userCount=0;

                            for (UserTable1 eligibleUser : eligibleUsers)
                            {
                                try
                                {
                                    cs.submit( 
                                            new Callable<User>()
                                            {
                                                public User call()
                                                {
                                                    int userId  = eligibleUser.getUserId();
                                                    List<EmployeeTable2> empData = db.findByUserId(userId);

                                                    EmployeeTable2 emp = null;

                                                    if (null != empData && !empData.isEmpty())
                                                    {
                                                        emp = empData.get(0);
                                                    }else{
                                                        String errorMsg = "No record found for given User ID in emp table";
                                                        logger.error(errorMsg);
                                                        //continue;
// conitnue does not work here. 
                                                    }           
                                                    User user = new User();
                                                    user.setUserId(userId);
                                                    user.setFullName(emp.getFullName());
                                                    return user;
                                                }
                                            }
                                        );
                                        userCount++;
                                }
                                catch(Exception ex)
                                {
                                    String errorMsg = "Error while creating map :" + ex.getMessage();
                                    logger.error(errorMsg);
                                }
                            }
                            for (int i = 0; i < userCount ; i++ ) {
                                try {
                                    User user = cs.take().get();
                                    if (user != null) {
                                        userIdMap.put(user.getUserId(), user);                                  
                                    }
                                } catch (ExecutionException e) {

                                } catch (InterruptedException e) {

                                }
                            }
                            executor.shutdown();
                            map.put(key, userIdMap);
                    }
                    catch(Exception ex)
                    {
                        String errorMsg = "Error while creating map :" + ex.getMessage();
                        logger.error(errorMsg);
                    }
                }
            }
            catch(Exception ex){
                String errorMsg = "Error while creating map :" + ex.getMessage();
                logger.error(errorMsg);
            }

            logger.info("Size of Map : " + map.size());
            Set<String> periods = map.keySet();
            logger.info("Size of periods : " + periods.size());
            for(String period :periods)
            {
                Map<Integer, User>  mapOfuserIds = map.get(period);
                Set<Integer> userIds = mapOfuserIds.keySet();
                logger.info("Size of Set : " + userIds.size());
                for(Integer userId : userIds){
                    User inf = mapOfuserIds.get(userId);
                    logger.info("User Id : " + inf.getUserId());
                }
            }
             long endTime = System.currentTimeMillis();
             long timeTaken = (endTime - startTime);
             logger.info("All threads are completed in " + timeTaken + " milisecond");
            logger.info("******END******");
        }

Answer 1

You really don't want to create a thread pool with as many threads as users you've read from the db. That doesn't make sense most of the time because you need to keep in mind that threads need to run somewhere... There are not many servers out there with 10 or 100 or even 1000 cores reserved for your application. A much smaller value like maybe 5 is often enough, depending on your environment.

And as always for topics about performance: You first need to test what your actual bottleneck is. Your application may simply don't benefit of threading because e.g. you are reading form a db which only allows 5 concurrent connections a the same time. In that case all your other 995 threads will simply wait.

Some other thing to consider is network latency: Reading multiple user ids from multiple threads may even increase the round trip time needed to get the data for one user from the database. An alternative approach might be to not read one user at a time, but the data of all 10'000 of them at once. That way your maybe available 10 GBit Ethernet connection to your database might really speed things up because you have only small communication overhead with the database but it might serve you all data you need in one answer quickly.

So in short, from my opinion your question is about performance optimization of your problem in general, but you don't know enough yet to decide which way to go.

Answer 2

you could try something like that:

List<String> periods = db.getPeriods(todayDate);
Map<String, Map<Integer, User>> hm = new HashMap<>();

periods.parallelStream().forEach(s -> {
   eligibleUsers = // getEligibleUsers();
   hm.put(s, eligibleUsers.parallelStream().collect(
   Collectors.toMap(UserTable1::getId,createUserForId(UserTable1:getId))
  });
); // 

And in the createUserForId you do your db-reading

private User createUserForId(Integer id){
        db.findByUserId(id);

        //...

         User user = new User();
        user.setUserId(userId);
        user.setFullName(emp.getFullName());
        return user;
    }