Chinaunix首页 | 论坛 | 博客

厚德载物peng.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

chengxiaopeng

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

hello world.

文章分类

全部博文（308）

最近访客
推荐博文
相关博文
java生产者消费者模式，不间断扫表，多线程处理数据，封装泛型类

分类： Java

2016-10-25 10:27:21

    在工作中，我们有时候需要对表，甚至是多个表，进行循环扫描，如果含有数据进行数据处理。这个让人很容易想到用到，生产者消费者模式，进行数据处理。如果具体的编程时你会发现，网上的很多例子是在单生产者多消费者模式下，正常运行，如果出现了多个生产者时，则会造成阻塞，或者有一个另外一个生产者的数据，无法及时的处理。
    仔细阅读代码，你会发现问题在于，当多个生产者，监控到缓冲区中没有数据，而临界资源访问时，只能是单个线程进行。这样会造成一个生产者进入临界资源，而另外的生成者线程被阻塞，当访问临界资源的那个生产者，结束访问临界资源后，缓冲区已经有了资源,这个时候，按照逻辑，另外的生成者则被阻塞。在极端的情况下，则会造成多个生产者，一直被阻塞无法进入缓冲区，无法进行数据处理。
    明白问题所在，一般就知道了解决的思路。问题在于多个生产者竞争访问临界资源。我采用的思路是，将多个生成者转化为一个生产者，用一个生产者访问临界资源。通过观察者模式，多需要生产数据时，遍历各个生成者，将数据汇总后，一次送入缓冲区。另外设置监听线程，只要缓冲区没有数据，即通过观察者模式，通知生成者生产数据（也就是扫表操作）。为方便使用，采用泛型类，泛型接口。方便系统接入。代码如下：

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer;

  3. import java.util.ArrayList;
  4. import java.util.LinkedList;
  5. import java.util.List;

  7. import org.apache.log4j.Logger;

  9. //生产者，消费者所使用的缓冲区
  10. //author:程晓鹏
  11. //date:2016.10.17
  12. public class BufferStorage<T>{
  13.     
  14.     private static final Logger log = Logger.getLogger(BufferStorage.class);
  15.     private LinkedList<T> bufferList = new LinkedList<T>(); //缓冲区集合
  16.     
  17.      //缓冲区中的产品个数
  18.      public int bufferSize(){
  19.          synchronized (this.bufferList){
  20.              return this.bufferList.size();
  21.          }
  22.      }
  23.     
  24.     //生产者，生产产品
  25.     public void produce(List<T> t){
  26.         synchronized (this.bufferList){
  27.             log.info("【缓冲区】生产者的产品，进入缓冲区，数量:" + t.size());
  28.             while(this.bufferList.size() > 0){ //当缓冲区中有数据时，暂停生产
  29.                 try{
  30.                     bufferList.wait(); //生产阻塞
  31.                 }catch (InterruptedException e){
  32.                     e.printStackTrace();
  33.                 }
  34.             }
  35.             
  36.             log.info("【缓冲区】产品开始添加到缓冲区");
  37.             for(int i=0; i<t.size(); i++){
  38.                 bufferList.add(t.get(i));
  39.             }
  40.             log.info("【缓冲区】产品添加完成 size=" + t.size() + "\n");            
  41.             bufferList.notifyAll();
  42.         }
  43.     }
  44.     
  45.     //消费者，消费产品
  46.     public T consume(){    
  47.         synchronized (this.bufferList){
  48.             while(this.bufferList.size() == 0){ //当缓冲区，没有产品时，消费者被阻塞
  49.                 try {
  50.                     this.bufferList.wait();    
  51.                 } catch (InterruptedException e) {
  52.                     e.printStackTrace();
  53.                 }
  54.             }
  55.             
  56.             T result = this.bufferList.getFirst();
  57.             this.bufferList.remove();
  58.             this.bufferList.notifyAll();
  59.             return result;
  60.         }
  61.         
  62.     }
  63.     
  64. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer;

  3. import java.util.List;

  5. //生产，观察者接口
  6. //author:程晓鹏
  7. //date:2016.10.17
  8. public interface IWatcherProducer<T> {
  9.     //开始生产产品
  10.     public List<T> beginProduce(String msg);
  11. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer;

  3. import java.util.List;
  4. import org.apache.log4j.Logger;

  6. //产品生产者类
  7. //author:程晓鹏
  8. //date:2016.10.17
  9. public abstract class ProducerThread<T> extends Thread implements IWatcherProducer<T>{
  10.     
  11.     private static final Logger log = Logger.getLogger(ProducerThread.class);

  13.      //默认构造函数
  14.      public ProducerThread(){
  15.         
  16.      }
  17.     
  18.      //开始运行
  19.      public void run(){
  20.         
  21.      }
  22.     
  23.      //实现接口
  24.      public List<T> beginProduce(String msg){
  25.          log.info("【生产者 thread_id="+this.getId()+"】收到管理线程发送的消息：" + msg);        
  26.          List<T> productData = this.produce();
  27.          int size = productData!=null ? productData.size() : 0;
  28.          log.info("【生产者】生产产品数量：" + size + " thread_id：" + this.getId());
  29.          return productData;
  30.      }
  31.     
  32.      //抽象生产者生成产品
  33.      public abstract List<T> produce();    
  34. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer;

  3. import org.apache.log4j.Logger;

  5. //消费者类
  6. //author:程晓鹏
  7. //date:2016.10.17
  8. public abstract class ConsumerThread<T> extends Thread{
  9.     private static final Logger log = Logger.getLogger(ConsumerThread.class);
  10.     private BufferStorage<T> buffstorage; //缓冲区仓库
  11.     private long lthreadId = this.getId(); //线程id
  12.     private long sleepTime;
  13.     
  14.      //构造函数
  15.      public ConsumerThread(BufferStorage<T> storage){
  16.          this.buffstorage = storage;
  17.          sleepTime = 100;
  18.      }
  19.     
  20.      public void run(){
  21.          this.beginConsume();        
  22.      }
  23.     
  24.      private void beginConsume(){
  25.          while(true){
  26.              log.info("【消费者】缓冲区，产品数：" + this.buffstorage.bufferSize() + " thread_id：" + lthreadId);
  27.              if(this.buffstorage.bufferSize() == 0){
  28.                  log.info("【消费者】缓冲区无产品，消费者线程休眠 " + sleepTime + "ms.");
  29.                  try {
  30.                     sleep(sleepTime);
  31.                  } catch (InterruptedException e) {
  32.                     // TODO Auto-generated catch block
  33.                     e.printStackTrace();
  34.                  }                
  35.              }
  36.             
  37.              T t = this.buffstorage.consume();
  38.              if(t != null){
  39.                  this.consume(t);
  40.              }
  41.          }
  42.      }
  43.     
  44.      //消费者，消费产品
  45.      public abstract void consume(T t);
  46. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer;

  3. import java.util.ArrayList;
  4. import java.util.List;
  5. import org.apache.log4j.Logger;

  7. //线程管理类
  8. public class ManageThread<T> extends Thread{
  9.     private static final Logger log = Logger.getLogger(ManageThread.class);
  10.     private List<IWatcherProducer<T>> listProducer = new ArrayList<IWatcherProducer<T>>(); //生产者观察者集合
  11.     private List<T> productDataList = new ArrayList<T>(); //生产者生产的产品集合
  12.     private BufferStorage<T> buffstorage; //缓冲区仓库
  13.     private long sleepTime;
  14.     
  15.     //构造函数
  16.      public ManageThread(BufferStorage<T> storage){
  17.          this.buffstorage = storage;
  18.          sleepTime = 1000;
  19.      }
  20.     
  21.     //注册生产观察者对象
  22.      public void attachProducer(IWatcherProducer<T> observer){        
  23.          this.listProducer.add(observer);
  24.      }
  25.     
  26.      //删除生产观察者对象
  27.      public void detachProducer(IWatcherProducer<T> observer){
  28.          this.listProducer.remove(observer);
  29.      }
  30.     
  31.      //通知所有注册的生产观察者对象
  32.      private void nodifyProducerObservers(String msg){
  33.          synchronized(this.productDataList){
  34.              int len = this.listProducer.size();
  35.              for(int i=0; i<len; i++){
  36.                  IWatcherProducer<T> observer = this.listProducer.get(i);
  37.                  log.info(String.format("【管理线程】" + msg + " --> 进度 %d/%d", i+1, len));
  38.                  List<T> data = observer.beginProduce(msg);
  39.                  if(null != data){
  40.                      for(int j=0; j<data.size(); j++){
  41.                          productDataList.add(data.get(j));
  42.                      }
  43.                  }
  44.              }
  45.              log.info("【管理线程】通知所有已注册的生产者 size:"+this.productDataList.size()+" finish.\n");
  46.          }
  47.      }
  48.     
  49.      public void run(){
  50.          while(true){
  51.              this.beginRun();
  52.              try {
  53.                     this.sleep(sleepTime);
  54.                 } catch (InterruptedException e) {
  55.                     // TODO Auto-generated catch block
  56.                     e.printStackTrace();
  57.                 }
  58.          }
  59.      }
  60.     
  61.      //开始运行
  62.      private void beginRun(){
  63.          log.info("【管理线程】缓冲区，产品数：" + this.buffstorage.bufferSize() + " thread_id：" +this.getId());
  64.          if(this.buffstorage.bufferSize() == 0){
  65.              this.nodifyProducerObservers("缓冲区中没有产品，生产者需要生产产品.");
  66.              if(this.productDataList.size() > 0){
  67.              this.buffstorage.produce(this.productDataList); //送入缓冲区
  68.              this.productDataList.clear(); //清空数据
  69.          }
  70.          }
  71.      }
  72. }
测试程序如下：

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer.demo;

  3. import java.util.ArrayList;
  4. import java.util.List;


  7. import com.hualong.entity.ResponseXml;
  8. import com.hualong.tux.CallLua;
  9. import com.hualong.tux.SmsData;
  10. import com.huawei.G4Data.Java2Xml;
  11. import com.common.ProducerConsumer.BufferStorage;
  12. import com.common.ProducerConsumer.ConsumerThread;

  14. //短信消费者
  15. public class SmsConsumer<SmsData> extends ConsumerThread<com.hualong.tux.SmsData>{

  17.     public SmsConsumer(BufferStorage<com.hualong.tux.SmsData> storage) {
  18.         super(storage);
  19.     }
  20.     
  21.     //消费者
  22.     public void consume(com.hualong.tux.SmsData t)
  23.     {
  24.         Java2Xml myxml = new Java2Xml();
  25.         ResponseXml result = CallLua.callBsh(t, "cxll_4g.bsh");
  26.         myxml.loadJavaObject(result);
  27.         String str ="";
  28.         String strXml = myxml.getXmlResult();
  29.         System.out.println(strXml); //打印结果
  30.     }
  31. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer.demo;

  3. import java.util.ArrayList;
  4. import java.util.List;

  6. import com.hualong.tux.SmsData;
  7. import com.common.ProducerConsumer.BufferStorage;
  8. import com.common.ProducerConsumer.ProducerThread;

  10. public class SmsProducer<SmsData> extends ProducerThread<com.hualong.tux.SmsData> {
  11.     
  12.     private int i_max_request = 2;
  13.     private int i_count = 0;
  14.     
  15.     //构造函数
  16.     public SmsProducer() {
  17.         super();
  18.     }
  19.     
  20.     public List<com.hualong.tux.SmsData> produce()
  21.     {
  22.         List<com.hualong.tux.SmsData> listArray = new ArrayList<com.hualong.tux.SmsData>();
  23.         if(i_count < i_max_request){ //模拟正在请求时，没有数据的情况
  24.             for(int i=0; i<5; i++){
  25.                 com.hualong.tux.SmsData sms = new com.hualong.tux.SmsData();
  26.                 sms.setRowid("AAAV6cAAGAAM3S9AAB");
  27.                 sms.setSmsId("MO_0934410000271A_20161014093441");
  28.                 sms.setUserNum("66666666666");
  29.                 sms.setCityCode("0372");
  30.                 sms.setMessage("0760ABC11");
  31.                 
  32.                 listArray.add(sms);
  33.             }
  34.             i_count++;
  35.         }
  36.         return listArray;
  37.         
  38.     }
  39. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer.demo;

  3. import java.util.ArrayList;
  4. import java.util.List;

  6. import com.hualong.tux.SmsData;
  7. import com.common.ProducerConsumer.BufferStorage;
  8. import com.common.ProducerConsumer.ProducerThread;

  10. public class SmsProducer2<SmsData> extends ProducerThread<com.hualong.tux.SmsData> {
  11.     
  12.     private int i_max_request = 2;
  13.     private int i_count = 0;
  14.     
  15.     //构造函数
  16.     public SmsProducer2(){
  17.         super();
  18.     }
  19.     
  20.     public List<com.hualong.tux.SmsData> produce()
  21.     {
  22.         List<com.hualong.tux.SmsData> listArray = new ArrayList<com.hualong.tux.SmsData>();
  23.         if(i_count < i_max_request){ //模拟正在请求时，没有数据的情况
  24.             for(int i=0; i<6; i++){
  25.                 com.hualong.tux.SmsData sms = new com.hualong.tux.SmsData();
  26.                 sms.setRowid("AAAV6cAAGAAM3S9AAB");
  27.                 sms.setSmsId("MO_0934410000271A_20161014093441");
  28.                 sms.setUserNum("18639551269");
  29.                 sms.setCityCode("0371");
  30.                 sms.setMessage("07600011");
  31.                 
  32.                 listArray.add(sms);
  33.             }
  34.             i_count++;
  35.         }
  36.         return listArray;
  37.         
  38.     }
  39. }

点击(此处)折叠或打开

  1. package com.common.ProducerConsumer.demo;

  3. import org.apache.log4j.PropertyConfigurator;

  5. import com.hualong.tux.SmsData;
  6. import com.common.ProducerConsumer.BufferStorage;
  7. import com.common.ProducerConsumer.ManageThread;

  9. public class testclient {
  10.     public static void main(String[] args) {
  11.         BufferStorage<SmsData> buffer = new BufferStorage<SmsData>();
  12.         ManageThread<SmsData> manager = new ManageThread<SmsData>(buffer);
  13.         SmsProducer<SmsData> producer = new SmsProducer<SmsData>();
  14.         SmsProducer2<SmsData> producer2 = new SmsProducer2<SmsData>();
  15.         
  16.         manager.attachProducer(producer); //注册生产者
  17.         manager.attachProducer(producer2);//注册生产者
  18.         
  19.         SmsConsumer<SmsData> consumer = new SmsConsumer<SmsData>(buffer);
  20.         SmsConsumer<SmsData> consumer2 = new SmsConsumer<SmsData>(buffer);
  21.         SmsConsumer<SmsData> consumer3 = new SmsConsumer<SmsData>(buffer);
  22.         
  23.         //log4j配置文件地址
  24.         String log4j_url = "/home/path/log4j_ProducerConsumer.properties";
  25.         PropertyConfigurator.configure(log4j_url);
  26.         
  27.         manager.start();
  28.         producer.start();
  29.         producer2.start();
  30.         
  31.         consumer.start();
  32.         consumer2.start();
  33.         consumer3.start();
  34.     }
  35. }

点击(此处)折叠或打开

  1. log4j.rootLogger = info,default

  3. log4j.appender.console = org.apache.log4j.ConsoleAppender
  4. log4j.appender.console.Target = System.out
  5. log4j.appender.console.layout = org.apache.log4j.PatternLayout
  6. log4j.appender.console.layout.ConversionPattern = %p %d{HH:mm:ss} %l %m %n

  8. log4j.appender.default = org.apache.log4j.DailyRollingFileAppender
  9. log4j.appender.default.File = /home/smsapp/Sms4gClient/logs/sms4g.log
  10. log4j.appender.default.Append = true
  11. log4j.appender.default.Threshold = info
  12. log4j.appender.default.layout = org.apache.log4j.PatternLayout
  13. log4j.appender.default.layout.ConversionPattern = %p %d{MM-dd HH:mm:ss} %l %m %n

  15. #class print #####################################################################
  16. log4j.logger.com.huawei.ProducerConsumer.BufferStorage=info,buffer
  17. log4j.appender.buffer = org.apache.log4j.DailyRollingFileAppender
  18. log4j.appender.buffer.File = /home/smsapp/Sms4gClient/logs/buffer.log
  19. log4j.appender.buffer.Append = true
  20. log4j.appender.buffer.Threshold = info
  21. log4j.appender.buffer.layout = org.apache.log4j.PatternLayout
  22. log4j.appender.buffer.layout.ConversionPattern = %d{MM-dd HH:mm:ss} %m %n

  24. log4j.logger.com.huawei.ProducerConsumer.ConsumerThread=info,Consumer
  25. log4j.appender.Consumer = org.apache.log4j.DailyRollingFileAppender
  26. log4j.appender.Consumer.File = /home/smsapp/Sms4gClient/logs/consumer.log
  27. log4j.appender.Consumer.Append = true
  28. log4j.appender.Consumer.Threshold = info
  29. log4j.appender.Consumer.layout = org.apache.log4j.PatternLayout
  30. log4j.appender.Consumer.layout.ConversionPattern = %d{MM-dd HH:mm:ss,SSS} %m %n

  32. log4j.logger.com.huawei.ProducerConsumer.ManageThread=info,listener
  33. log4j.appender.listener = org.apache.log4j.DailyRollingFileAppender
  34. log4j.appender.listener.File = /home/smsapp/Sms4gClient/logs/listener.log
  35. log4j.appender.listener.Append = true
  36. log4j.appender.listener.Threshold = info
  37. log4j.appender.listener.layout = org.apache.log4j.PatternLayout
  38. log4j.appender.listener.layout.ConversionPattern = %d{MM-dd HH:mm:ss} %m %n

  40. log4j.logger.com.huawei.ProducerConsumer.ProducerThread=info,producer
  41. log4j.appender.producer = org.apache.log4j.DailyRollingFileAppender
  42. log4j.appender.producer.File = /home/smsapp/Sms4gClient/logs/producer.log
  43. log4j.appender.producer.Append = true
  44. log4j.appender.producer.Threshold = info
  45. log4j.appender.producer.layout = org.apache.log4j.PatternLayout
  46. log4j.appender.producer.layout.ConversionPattern = %d{MM-dd HH:mm:ss} %m %n
ProducerConsumer.rar


阅读(4022) | 评论(0) | 转发(0) |
0

上一篇：处理数据文件，对文件最后一行没有结束符，追加结束符

下一篇：个性化的nohup，可以判断程序是否启动，如启动，则不重复启动指定程序

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

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

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

16024965号-6