java 使用多线程批处理问题

0 0

java 使用多线程批处理问题5

我的需求是我需要处理一批数据，这些数据放到一个List 里面。单线程是这样的
while(list不为空){
process(list.next());
}

这时发现，每执行一次process需要5分钟。效率太低了。

我希望能够多线程实现批处理。比如每次开10个线程同时执行process（），每一个线程处理一组数据。
while(list不为空){

process(list.next());
process(list.next());
process(list.next());
process(list.next());
.
.
.
.
}

这时发现，这不只是开启了一个线程，而是为每一组数据开了一个线程，电脑崩溃了。

我希望开启线程的数量是10，而当某一个线程执行完一个process（）之后。再执行另外一个process处理另外一组数据。如何实现呢？

Java 多线程

2013年5月12日 23:21

zjwlonger
1
0 0 3

4个答案按时间排序按投票排序

0 0

你的崩溃是不是多线程同时遍历和操作这个list产生的啊？我比较好奇的是什么什么过程需要执行5分钟……没见过这样的，真心求教

2013年5月13日 16:29

kevinpan45
5
0 0 12

添加评论

0 0

1、你可以采用批量保存，hibernate的实现http://oss.org.cn/ossdocs/framework/hibernate/reference-v3_zh-cn/batch.html

2、如果你要采用多线程保存数据，要注意主键生成策略不要是increment。
java中有个ThreadPoolExecutor，你可以定义线程池中线程的数量，将保存的数据当成task，用线程来处理

2013年5月13日 09:39

Java_Eye007
44
0 0 4

1条评论

0 0

    package com.kuke.auth.log.controller;  
      
    import java.util.ArrayList;  
    import java.util.List;  
    import java.util.Map;  
    import java.util.Vector;  
    import java.util.concurrent.ExecutorService;  
    import java.util.concurrent.Executors;  
      
    import javax.servlet.http.HttpServletRequest;  
    import javax.servlet.http.HttpServletResponse;  
      
    import org.springframework.beans.factory.annotation.Autowired;  
    import org.springframework.stereotype.Controller;  
    import org.springframework.web.bind.annotation.RequestMapping;  
      
    import com.kuke.auth.regist.domain.UserActionLog;  
    import com.kuke.auth.regist.domain.UserListenLog;  
    import com.kuke.auth.regist.service.RegistServiceImpl;  
    import com.kuke.core.base.action.AbstractActionBean;  
      
    @Controller  
    @RequestMapping("/kuke/log")  
    public class LogController extends AbstractActionBean {  
      
        private static ExecutorService pool = Executors.newCachedThreadPool();//线程池  
      
        private static List<UserActionLog> userActionList = new ArrayList<UserActionLog>();  
      
        private static List<UserListenLog> userListenList = new ArrayList<UserListenLog>();  
      
        private static List<UserActionLog> tempList;  
      
        private static List<UserListenLog> tempList1;  
      
        @Autowired  
        private RegistServiceImpl registServiceImpl;  
      
        /** 
         *  
         * action log 
         */  
        @RequestMapping(value = "/userAction")  
        public synchronized void userActionLog(HttpServletRequest request,  
                HttpServletResponse response) throws Exception {  
            String userId = this.getLoginUser().getUid();  
      
            Map<String, String> params = getParameterMap(request);  
            String requestIp = params.get("operator_ip");  
            String actionType = params.get("actionType");  
            String start_date = params.get("start_date");  
            String end_date = params.get("end_date");  
      
            UserActionLog useractionlog = new UserActionLog();  
            // useractionlog.setId(IdGenerator.getUUIDHex32());  
            useractionlog.setAction_type(actionType);  
            useractionlog.setUser_id(userId);  
            useractionlog.setStart_state(start_date);  
            useractionlog.setEnd_state(end_date);  
            useractionlog.setChannel_type("");  
            useractionlog.setOperator_ip(requestIp);  
            userActionList.add(useractionlog);  
            if (userActionList.size() == 100) {  
    //          System.out.println("=======1======"+userActionList.size());  
                tempList = new ArrayList<UserActionLog>();  
                tempList.addAll(userActionList);  
                pool.execute(new Runnable() {  
                    public void run() {  
                        try {  
                            registServiceImpl.addUseractionlog(tempList);  
                        } catch (Exception e) {  
                            e.printStackTrace();  
                        }  
                    }  
                });  
                 userActionList.clear();  
                userActionList = new ArrayList<UserActionLog>();  
            }  
        }  
      
        /** 
         * listen log 
         *  
         */  
        @RequestMapping(value = "/userListen")  
        public synchronized void userListenLog(HttpServletRequest request,  
                HttpServletResponse response) throws Exception {  
            String userId = this.getLoginUser().getUid();  
      
            Map<String, String> params = getParameterMap(request);  
            String requestIp = params.get("operator_ip");  
            String l_code = params.get("l_code");  
            String listen_type = params.get("listen_type");  
            String listen_date = params.get("listen_date");  
            String item_code = params.get("item_code");  
            String from_client = params.get("from_client");  
            String cname = params.get("cname");  
            String ename = params.get("ename");  
            String listen_ip = params.get("listen_ip");  
            String channel_type = params.get("channel_type");  
            String org_user_id = params.get("org_user_id");  
            String ischarged = params.get("ischarged");  
      
            UserListenLog userListenLog = new UserListenLog();  
            userListenLog.setListen_ip(requestIp);  
            userListenLog.setL_code(l_code);  
            userListenLog.setListen_type(listen_type);  
            userListenLog.setListen_type(l_code);  
            userListenLog.setListen_date(listen_date);  
            userListenLog.setItem_code(item_code);  
            userListenLog.setFrom_client(from_client);  
            userListenLog.setCname(cname);  
            userListenLog.setEname(ename);  
            userListenLog.setListen_ip(listen_ip);  
            userListenLog.setChannel_type(channel_type);  
            userListenLog.setOrg_user_id(org_user_id);  
            userListenLog.setIscharged(ischarged);  
            userListenLog.setListen_user_id(userId);  
            userListenList.add(userListenLog);  
            if (userListenList.size() == 100) {  
                System.out.println("=======11======"+userListenList.size());  
                tempList1 = new ArrayList<UserListenLog>();  
                tempList1.addAll(userListenList);  
                pool.execute(new Runnable() {  
                    public void run() {  
                        try {  
                            registServiceImpl.addUserListenLog(tempList1);  
                        } catch (Exception e) {  
                            e.printStackTrace();  
                        }  
                    }  
                });  
                userListenList.clear();  
                userListenList = new ArrayList<UserListenLog>();  
            }  
      
        }  
    }

2013年5月13日 09:34