Chinaunix首页 | 论坛 | 博客

厚德载物peng.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

chengxiaopeng

  • 博客访问： 2538807
  • 博文数量： 308
  • 博客积分： 5547
  • 博客等级： 大校
  • 技术积分： 3782
  • 用 户 组： 普通用户
  • 注册时间： 2009-11-24 09:47
个人简介

hello world.

文章分类

全部博文（308）

最近访客
推荐博文
相关博文
使用生产者，消费者模式，进行大数据量的处理(java)

分类： Java

2013-04-19 16:11:59

    如果你搜索，生产者，消费者模式。你会看到很多的原理，例子等。而这些例子，大部分都有一个共同特征，那就是，这个程序是不停的在进行运行，没有停止下来。当然了，这方便教学。让人能很方便，快捷的明白生产者，消费者程序模式的原理。但在具体的工作中，似乎不太常用。
因为一般在工作中，生产者的数量是固定的，程序总要运行完，总要停止。我在工作中，就遇到这样的情况。
    因为工作需要，我需要给另外一个系统提供全量数据，经过初期的数据整理工作，全量数据达到400多万的数据量，但还有一些数据项，在另外的系统中，我只能通过WebService调用其他系统，以获取我需要的数据，才能最终完成数据工作。这也就意味着，每一条数据我都要进行一次请求另外的系统。因为这400多万的数据量。你可以用简单的方法，如select * from tbA，可400多万的数据量(也许你处理的数据量，比我的还要大)，这样搞，很可能内存会溢出。这时候，大家一般都会想到使用分块处理，比如说一次读2万条，处理完后，然后再继续读取，处理，直到数据处理完。这个时候，在这里我采用一个生产者，多个消费者的方式。
    选定这种处理方式后，肯定需要对已经处理过的数据，做一个标识。这样就可以区分，已经完成的，未完成的。因此我在原来的表结构上追加一个字段 deal_tag。默认值是0。
    处理的方式如下：
    生产者读取指定的数据记录数(假设为2万)。
    1.更新rownum<=20000的数据，将其标识位更新为'B'。此部操作的目的是，选取要处理的数据。
    2.查询标志位为'B'的数据，将其读到缓冲区中，更新原有的'B'标志位'BB'标识此数据已经进入缓冲区。
    3.此时消费者开始对进入缓冲区的数据，进行处理。当缓冲区的数据处理完后，对通知生产者继续读数据，如果要进行处理的数据读完后，生产者，更新一个生产者完成的标志。此时生产者线程正常结束。消费者结束的标志为：生产完成的标志为true，并且缓冲区中也没有数据，消费者也就结束。当消费者线程都完成后，则程序也就结束了。
     是不是，说的你也迷糊了。那先把代码放上来(资源管理)：

点击(此处)折叠或打开

  1. package com.epro.client;


  4. import java.util.ArrayList;
  5. import java.util.List;
  6. import org.apache.log4j.Logger;
  7. import com.epro.entity.FaultInfo;


  10. /**
  11.  * objcect: TradeManage
  12.  * author 程晓鹏
  13.  * description: 工单管理类
  14.  * date Mar 19, 2013 3:56:33 PM
  15.  */
  16. public class TradeManage {
  17.     
  18.     private List<FaultInfo> tradeList; //工单数据列表
  19.     private static Logger log = Logger.getLogger(TradeManage.class);
  20.     public boolean readOK;
  21.     
  22.     /**
  23.      * 默认构造函数
  24.      */
  25.     public TradeManage(){
  26.         tradeList = new ArrayList<FaultInfo>();
  27.         readOK = false;
  28.     }
  29.     
  30.     
  31.     /**
  32.      * 进行资源生产
  33.      * @param listProducer 生产的资源集合
  34.      */
  35.     public synchronized void increace(List<FaultInfo> listProducer){
  36.         while(tradeList.size() != 0){
  37.             try{
  38.                 wait();
  39.             }catch (InterruptedException e) {
  40.                 e.printStackTrace();
  41.                 log.error("increace error: " + e.getMessage());
  42.             }
  43.         }
  44.         if(listProducer.size() > 0){
  45.             this.tradeList.addAll(listProducer);            
  46.         }else{
  47.             readOK = true; //没有往缓冲区中放数据，说明读取操作完成
  48.         }
  49.         
  50.         this.notifyAll();
  51.     }
  52.     
  53.     /**
  54.      * 消费者进行资源消费
  55.      * @return FaultInfo对象
  56.      */
  57.     public synchronized FaultInfo decreace(){
  58.         FaultInfo result = null;
  59.         while(tradeList.size() == 0){
  60.             if(!readOK){
  61.                 try{
  62.      wait();
  63.                 }
  64.                 catch (InterruptedException e) {
  65.      e.printStackTrace();
  66.      log.error("decreace error: " + e.getMessage());
  67.      }
  68.             }else{
  69.                 break;
  70.             }
  71.         }
  72.         
  73.         if(tradeList.size() > 0){
  74.             result = tradeList.remove(0);
  75.         }
  76.         this.notifyAll();
  77.         return result;
  78.     }
  79. }
生产者的代码，生产者，主要进行数据的读完，并放入到缓冲区中：

点击(此处)折叠或打开

  1. package com.epro.client;

  3. import java.sql.ResultSet;
  4. import java.sql.SQLException;
  5. import java.util.ArrayList;
  6. import java.util.List;
  7. import java.util.regex.Matcher;
  8. import java.util.regex.Pattern;

  10. import org.apache.log4j.Logger;

  12. import com.epro.DBUtility.DBHelper;
  13. import com.epro.entity.FaultInfo;

  15. /**
  16.  * objcect: ReadThread
  17.  * author 程晓鹏
  18.  * description: 进行数据读取的线程
  19.  * date Mar 19, 2013 4:20:45 PM
  20.  */
  21. public class ReadThread implements Runnable{
  22.     
  23.     private TradeManage manage;
  24.     public static String masterTableName = "ti_all_112_disk2";
  25.     private int readSize;
  26.     private DBHelper db;
  27.     private int totalReadSize = 0;
  28.     private static Logger log = Logger.getLogger(ReadThread.class);
  29.     /**
  30.      * 默认构造函数
  31.      */
  32.     public ReadThread(TradeManage trade){
  33.         manage = trade;
  34.         db = new DBHelper();
  35.         readSize = 0;
  36.         totalReadSize = 0;
  37.     }
  38.     
  39.     public void run(){
  40.         do{
  41.             manage.increace(this.readData(50000));            
  42.         }while(readSize > 0);
  43.     }

  45.     /**
  46.      * 进行数据读取
  47.      * @param readCount 读取的数据量
  48.      * @return 返回读取的数据数
  49.      */
  50.     public List<FaultInfo> readData(int readCount){
  51.         List<FaultInfo> result = new ArrayList<FaultInfo>();
  52.         String strSql = "";
  53.         this.dealOK();    //对上一批处理过的数据，进行提交            
  54.         strSql = getInitDataSql(readCount); //将要进行数据的处理的标志位更改为'B'
  55.         db.executeUpdate(strSql); //执行操作。
  56.         strSql = this.getReadSql(); //读取标志位为 'B'的数据
  57.         ResultSet rs = null;            
  58.         try {
  59.             rs = db.executeQuery(strSql);
  60.             while(rs.next()){
  61.                 FaultInfo item = new FaultInfo();
  62.                 item.CityCode = getString(rs.getString("eparchy_code"));
  63.                 item.Num = getString(rs.getString("serial_number"));
  64.                 item.ServiceID = getString(rs.getString("service_id"));
  65.                 item.InstallAddress = getString(rs.getString("detail_install_address"));
  66.                 item.bizCode = getString(rs.getString("product_class"));
  67.                 item.TownFlag = getString(rs.getString("town_flag"));
  68.                 item.Accounts = getString(rs.getString("product_id"));
  69.                 item.LineBaud = getString(rs.getString("rate"));
  70.                 item.Inmethod = getString(rs.getString("in_net_mode"));
  71.                 item.ContactPersonName = getString(rs.getString("link_name"));
  72.                 item.ContactPersonunit = getString(rs.getString("work_name"));
  73.                 item.ContactPersonPhone = getString(rs.getString("link_phone"));
  74.                 item.AcceptPhoneNum = getString(rs.getString("handset_num"));
  75.                 item.IdCardType = getString(rs.getString("pspt_type_code"));
  76.                 item.OperaterType = getString(rs.getString("operate_type"));
  77.                 item.IdCardNum = getString(rs.getString("pspt_id"));
  78.                 item.InTime = getString(rs.getString("in_time"));
  79.                 item.CustTypeCode = getString(rs.getString("cust_type_code"));
  80.                 item.CustTypeName = getString(rs.getString("cust_type_name"));
  81.                 item.UserID = getString(rs.getString("user_id"));
  82.                 item.NasCode = getString(rs.getString("nas_code"));
  83.                 item.NasName = getString(rs.getString("nas_name"));
  84.                 item.ServiceName = getString(rs.getString("service_name"));
  85.                 item.OfficeId = getString(rs.getString("moffice_id")); //局向编码
  86.                 item.OfficeName = getString(rs.getString("moffice_name")); //局向名称
  87.                 item.CustID = getString(rs.getString("cust_id"));
  88.                 item.CustName = getString(rs.getString("cust_name"));
  89.                 result.add(item); //放入集合中    
  90.             }
  91.             rs.close();
  92.         } catch (SQLException e) {
  93.             e.printStackTrace();
  94.             log.error(e.getMessage());
  95.         }finally{
  96.             rs = null;
  97.             db.Close();
  98.         }
  99.         readSize = result.size(); //获取读取的数量
  100.         if(readSize > 0){
  101.             totalReadSize += readSize;
  102.             log.info("read size: " + readSize + " total read size: " + totalReadSize);
  103.             strSql = this.getReadOKSql();
  104.             db.executeUpdate(strSql); //执行操作。    
  105.             log.info("read ok.");
  106.         }
  107.         return result;
  108.         
  109.     }    

  111.     /**
  112.      * 当数据处理完成后，进行标识为更改
  113.      */
  114.     public void dealOK(){
  115.         String strSql = "";
  116.         if(readSize > 0){
  117.             strSql = getDealOKSql(); //提交上一批数据，更改标志位为'OK'
  118.             db.executeUpdate(strSql); //执行操作。
  119.             log.info("commit size is: " + readSize);
  120.         }
  121.     }
  122.     
  123.     /**
  124.      * 对要入缓冲区的数据进行标志位更改
  125.      * @param readCount 读取的数据条数
  126.      * @return 对要进入缓冲区的数据进行标志位更改
  127.      */
  128.     private String getInitDataSql(int readCount){
  129.         StringBuilder result = new StringBuilder();
  130.         result.append("update " + masterTableName +" set deal_tag = 'B' where rownum <= ");
  131.         result.append(readCount);
  132.         result.append(" and deal_tag = '0'");
  133.         
  134.         return result.toString();
  135.     }
  136.     
  137.     private String getDealOKSql(){
  138.          return "update " + masterTableName +" set deal_tag = 'OK' where deal_tag = 'BB'";
  139.     }
  140.     
  141.     /**
  142.      * 读取完成的SQL语句
  143.      * @return SQL语句
  144.      */
  145.     private String getReadOKSql(){
  146.         return "update " + masterTableName +" set deal_tag = 'BB' where deal_tag = 'B'";
  147.     }
  148.     
  149.     /**
  150.      * 读取数据的SQL
  151.      * @return SQL字符串
  152.      */
  153.     private String getReadSql(){
  154.         StringBuilder result = new StringBuilder();
  155.         result.append("select tb1.eparchy_code, tb1.serial_number, tb1.service_id, ");
  156.         result.append(" tb1.detail_install_address, tb1.product_class, tb1.town_flag,");
  157.         result.append(" tb1.product_id, tb1.rate, tb1.in_net_mode,");
  158.         result.append(" nvl(tb1.link_name, '') link_name, tb1.work_name, ");
  159.         result.append(" nvl(tb1.link_phone, '') link_phone, tb1.handset_num, tb1.pspt_type_code, ");
  160.         result.append(" tb1.operate_type, tb1.pspt_id, ");
  161.         result.append(" to_char(in_time, 'yyyy-MM-dd hh24:mi:ss') in_time,");
  162.         result.append(" tb1.cust_type_code, ");
  163.         result.append(" decode(tb1.cust_type_name, ");
  164.         result.append(" null,decode(tb1.cust_type_code, ");
  165.         result.append(" '0', '个人客户',");
  166.         result.append(" '1', '集团客户', ''), ");
  167.         result.append(" tb1.cust_type_name) cust_type_name, tb1.user_id, ");
  168.         result.append(" tb1.nas_code, tb1.nas_name, tb1.service_name, ");
  169.         result.append(" tb1.moffice_id, tb1.moffice_name, "); //追加局向编码，局向名称
  170.         result.append(" tb1.cust_id, tb1.cust_name ");
  171.         result.append("from ");
  172.         result.append(masterTableName);
  173.         result.append(" tb1 ");
  174.         result.append("where tb1.deal_tag = 'B' ");
  175.         
  176.         return result.toString();
  177.     }
  178.     
  179.     //对null值进行屏蔽，返回空字符串
  180.     private String getString(String strInfo){
  181.         String result = "";
  182.         if( null != strInfo && !"".equals(strInfo)){
  183.             Pattern p = Pattern.compile("\t|\r|\n");
  184.             Matcher m = p.matcher(strInfo);
  185.             result = m.replaceAll(""); //将制表符，回车，换行符过滤掉
  186.         }
  187.         
  188.         return result;
  189.     }    
  190. }
消费者的代码，主要对进行缓冲区的数据，进行处理工作：

点击(此处)折叠或打开

  1. package com.epro.client;

  3. import java.io.File;
  4. import java.util.regex.Matcher;
  5. import java.util.regex.Pattern;

  7. import org.apache.log4j.Logger;
  8. import com.epro.Res.ResFacade;
  9. import com.epro.Res.ResInfo;
  10. import com.epro.entity.FaultInfo;

  12. /**
  13.  * 处理数据类
  14.  * @author 程晓鹏
  15.  *
  16.  */
  17. public class DealData implements Runnable {
  18.     
  19.     private GenerateFile genFile;
  20.     private ResFacade res;
  21.     private static Logger log = Logger.getLogger(DealData.class);
  22.     private TradeManage mytrade;
  23.     /**
  24.      * 默认构造函数
  25.      */
  26.     public DealData(TradeManage trade){
  27.         mytrade = trade;
  28.         res = new ResFacade();
  29.         genFile = new GenerateFile("result");    
  30.     }
  31.     
  32.     public void run() {    
  33.         this.beginDealData();        
  34.     }
  35.     
  36.     private void beginDealData(){
  37.         FaultInfo trade = null; //工单ID
  38.         do{
  39.             trade = mytrade.decreace();
  40.             if(trade != null){ //当没有数据时，跳出循环
  41.                 this.dealTrade(trade);
  42.             }
  43.         }while(trade != null); //当消费的资源为NULL时，则说明工作已经完成，可以跳出循环，结束线程
  44.     }
  45.     
  46.     private String getInstallAddress(String serviceName, String nasCode, String nasName){
  47.         StringBuilder result = new StringBuilder();
  48.         result.append(serviceName);
  49.         result.append(";");
  50.         result.append(nasCode);
  51.         result.append(";");
  52.         result.append(nasName);
  53.         
  54.         return result.toString();
  55.     }    
  56.     
  57.     //获取Num2的数据
  58.     private String getNum2(String cityCode, String num1)
  59.     {
  60.         String result = "";
  61.         if(num1.length() > 0 && num1.length() <=8){
  62.             result = cityCode + num1;
  63.         }
  64.         
  65.         return result;        
  66.     }
  67.     
  68.     private void dealTrade(FaultInfo trade){
  69.         String strTag = ""; //处理状态标志位
  70.         if("0".equals(trade.bizCode) || "1".equals(trade.bizCode)){
  71.             ResInfo myRes = res.getResInfo(trade.UserID);
  72.             if("0".equals(myRes.Result)){
  73.                 trade.Num1 = getString(myRes.RelaProductNo);
  74.                 trade.OfficeId = getString(myRes.OfficeCode);
  75.                 trade.OfficeName = getString(myRes.OfficeName);
  76.                 trade.DeviceCode = getString(myRes.LnFacCode);
  77.                 trade.LineType = getString(myRes.LineType);
  78.                 trade.CommitmentFlag = getString(myRes.CityFlag);
  79.             }else{
  80.                 trade.Num1 = "";
  81.                 trade.DeviceCode = "";
  82.                 trade.LineType = "";
  83.                 trade.CommitmentFlag = "";
  84.             }
  85.             trade.Num2 = getNum2(trade.CityCode, trade.Num1);        
  86.             strTag = myRes.Result;
  87.         }else if("2".equals(trade.bizCode)){
  88.             trade.Num1 = trade.Num;
  89.             trade.Num2 = trade.Num;
  90.             trade.InstallAddress = getInstallAddress(trade.ServiceName,trade.NasCode, trade.NasName);
  91.             trade.OfficeId = getString(trade.NasCode);
  92.             trade.OfficeName = getString(trade.NasName);
  93.             trade.LineType = "";
  94.             trade.DeviceCode = "";
  95.             trade.CommitmentFlag = "";
  96.             strTag = "0";
  97.         }        

  99.         if(!"0".equals(strTag)){
  100.             log.info("user_id: " + trade.UserID + " error_id: " +strTag);
  101.         }
  102.         
  103.         genFile.append(trade);    //追加数据
  104.     }
  105.     
  106.     
  107.     /**
  108.      * 获取生成的文件
  109.      * @return File对象
  110.      */
  111.     public File getGenerateFile(){
  112.         return genFile.getFile();
  113.     }
  114.     
  115.     //对null值进行屏蔽，返回空字符串
  116.     private String getString(String strInfo){
  117.         String result = "";
  118.         if( null != strInfo && !"".equals(strInfo)){
  119.             Pattern p = Pattern.compile("\t|\r|\n");
  120.             Matcher m = p.matcher(strInfo);
  121.             result = m.replaceAll(""); //将制表符，回车，换行符过滤掉
  122.         }
  123.         
  124.         return result;
  125.     }
  126. }
初始化，生产者，消费者的代码，就比较的简单了，我采用一个生产者，多个消费者的模式(因为读一次数据，就够消费者处理一阵子了)

点击(此处)折叠或打开

  1. /**
  2.      * 进行数据处理
  3.      */
  4.     private void dealData(){
  5.         TradeManage trade = new TradeManage();
  6.         
  7.         ReadThread read = new ReadThread(trade);
  8.         Thread myReadThread = new Thread(read);
  9.      log.info("master table: " + ReadThread.masterTableName );
  10.         myReadThread.start(); //启动读取数据库线程
  11.         
  12.         DealData mydeal = new DealData(trade);
  13.         List<Thread> listThread = new ArrayList<Thread>();
  14.         for(int i=0; i<100; i++){            
  15.             Thread myThread = new Thread(mydeal);
  16.             myThread.start(); //启动线程
  17.             listThread.add(myThread);
  18.         }
  19.         
  20.         while(listThread.size() > 0){
  21.             Thread mythread = listThread.get(0);
  22.             if("TERMINATED".equals(mythread.getState().toString())){
  23.                 listThread.remove(mythread);
  24.             }
  25.         }
  26.         read.dealOK(); //提交缓冲区数据，进行标志位更改
  27.         File genFile = mydeal.getGenerateFile(); //获取生成的文件
  28.         this.upFtpFile(genFile); //进行文件生成
  29.     }



阅读(8157) | 评论(1) | 转发(0) |
1

上一篇：使用CodeSmith进行三层架构代码，简单网页代码的生成

下一篇：使用AWK进行数据文件的检测

给主人留下些什么吧！~~

brokephp2013-11-20 15:10:51

正在寻找这方面的东西。。。谢啦

回复 | 举报
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6