浅谈JAVA设计模式

                  1
                public
                class
                 BackThreadCaller

                  2
                {

                  3
                  4
                private
                static
                volatile
                 BackThreadCaller instance;

                  5
                private
                 ThreadPoolExecutor pool;

                  6
                private
                static Logger log = LoggerFactory.getLogger(BackThreadCaller.class);
  7  8private BackThreadCaller()
  9    {
 10int corePoolSize = CalendarConsts.getThreadCorePoolSize();
 11int maximumPoolSize = CalendarConsts.getThreadMaximumPoolSize();
 12int keepAliveTime = CalendarConsts.getThreadKeepAliveTime();
 13int QueueSize = CalendarConsts.getThreadQueueSize();
 14         pool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS,
 15new ArrayBlockingQueue<Runnable>(QueueSize),
 16new ThreadPoolExecutor.DiscardPolicy());
 17    }
 18 19publicstatic BackThreadCaller getInstance()
 20    {
 21if (instance == null)
 22        {
 23synchronized (BackThreadCaller.class)
 24            {
 25if (instance == null)
 26                {
 27                     instance = new BackThreadCaller();
 28                }
 29            }
 30        }
 31         log.info("BackThreadCaller:ActiveCount={}|CompletedTaskCount={}|CorePoolSize={}|LargestPoolSize={}|PoolSize={}|TaskCount={}",
 32            instance.getActiveCount(),instance.getCompletedTaskCount(),
 33            instance.getCorePoolSize(),instance.getLargestPoolSize(),instance.getPoolSize(),
 34            instance.getTaskCount());
 35return instance;
 36    }
 37 38publicvoid startAThread(Thread thread)
 39    {
 40try 41        {
 42            pool.execute(thread);
 43        }
 44catch (Exception e)
 45        {
 46             log.error("Exception", e);
 47        }
 48    }
 49 50/** 51     * 开始执行一个线程
 52     * @param thread
 53*/ 54publicvoid startAThread(Runnable thread)
 55    {
 56try 57        {
 58            pool.execute(thread);
 59        }
 60catch (Exception e)
 61        {
 62             log.error("Exception", e);
 63        }
 64    }
 65 66// 正在执行的线程任务数,本例中如果为2,则可以观察两个office是否可以同时工作 67publicint getActiveCount()
 68    {
 69return pool.getActiveCount();
 70    }
 71 72// 返回池中的当前线程数 73publicint getPoolSize()
 74    {
 75return pool.getPoolSize();
 76    }
 77 78// 返回核心线程数 79publicint getCorePoolSize()
 80    {
 81return pool.getCorePoolSize();
 82    }
 83 84// 返回曾经同时位于池中的最大线程数 85publicint getLargestPoolSize()
 86    {
 87return pool.getLargestPoolSize();
 88    }
 89 90// 已经完成的任务总数 91publiclong getCompletedTaskCount()
 92    {
 93return pool.getCompletedTaskCount();
 94    }
 95 96// 曾经计划的任务总数 97publiclong getTaskCount()
 98    {
 99return pool.getTaskCount();
100    }
101102 }

                  1
                /**
                  2
                 * 

                  3
                 * 功能描述：日志上报工具类

                  4
                 *

                  5
                */
                  6
                public
                class
                 UActionTools

                  7
                {

                  8
                  9
                private
                static Logger log = LoggerFactory.getLogger(UActionTools.class);
 10privatestaticvolatile UActionTools instance;
 11 12privatefinal ConcurrentLinkedQueue<UserLog> uLogs = new ConcurrentLinkedQueue<UserLog>();
 13private ThreadPoolExecutor pool;
 14privateint delayTime = 60000;
 15privateint batchsize = 30;
 16 17/** 18     * 本方法拟采用批量提交的模式来实现.
 19     * 还有一种实现方法是：取消arraylist定义，每次把记录保存到pstmt,到一定数量后再executebatch.
 20     * 后一种方法需要数据库连接随时保持，占用一个连接，不太好。
 21*/ 22 23private UActionTools()
 24    {
 25         batchsize = Integer.valueOf(BComponetConfig.getInstance().getProperty(
 26             "Logsbatch", "30"));
 27         delayTime = Integer.valueOf(BComponetConfig.getInstance().getProperty(
 28             "LogsInterval", "60000"));
 29 30         pool = new ThreadPoolExecutor(8, 20, 30, TimeUnit.SECONDS,
 31new ArrayBlockingQueue<Runnable>(512),
 32new ThreadPoolExecutor.CallerRunsPolicy());
 33 34// 实例化线程对象后就立即实例化一个日志处理器并提交到线程池中
 35// 等待线程池调度，之后这个任务将一直运行直到线程池关闭 36         pool.execute(new LogHandler());
 37 38         log.info("start actionlog thread");
 39    }
 40 41publicstatic UActionTools getInstance()
 42    {
 43if (instance == null)
 44        {
 45synchronized (UActionTools.class)
 46            {
 47if (instance == null)
 48                {
 49                     instance = new UActionTools();
 50                }
 51            }
 52        }
 53return instance;
 54    }
 55 56/** 57     * 调用存储线程保存日志信息
 58     * 
 59     * @param ulog
 60*/ 61publicvoid saveAction(UserLog ulog)
 62    {
 63 64        uLogs.add(ulog);
 65 66    }
 67 68/** 69     * 获得当前内存日志信息
 70     * 
 71     * @return ConcurrentLinkedQueue<UserLog>
 72*/ 73public ConcurrentLinkedQueue<UserLog> getForceLists()
 74    {
 75return uLogs;
 76    }
 77 78// 正在执行的线程任务数,本例中如果为2,则可以观察两个office是否可以同时工作 79publicint getActiveCount()
 80    {
 81return pool.getActiveCount();
 82    }
 83 84// 返回池中的当前线程数 85publicint getPoolSize()
 86    {
 87return pool.getPoolSize();
 88    }
 89 90// 返回核心线程数 91publicint getCorePoolSize()
 92    {
 93return pool.getCorePoolSize();
 94    }
 95 96// 返回曾经同时位于池中的最大线程数 97publicint getLargestPoolSize()
 98    {
 99return pool.getLargestPoolSize();
100    }
101102// 已经完成的任务总数103publiclong getCompletedTaskCount()
104    {
105return pool.getCompletedTaskCount();
106    }
107108// 曾经计划的任务总数109publiclong getTaskCount()
110    {
111return pool.getTaskCount();
112    }
113114/**115     * 日志处理类
116     * 
117     * 依据生产者消费者模式来处理，多个生产者将数据放入队列，生产者将数据 放入队列就立即返回,
118     * logHandler当做消费者一直从队列中消费数据，再将数据生成一个个任务提交到线程池中去处理.
119     * 
120     * 
121     * 优点： 1.线程安全，不会丢失数据。 2.生成者放入数据到队列中就立即返回，生产者（主线程）的执行时间更短，之前的实现方式中
122     * 生产者（主线程）还需负责从队列中消费数据，然后才生产一个个任务提交到线程池中处理, 相比 之下缩短了主线程的处理时间，这样也更合理.
123     * 
124     * @author zhouguiping
125     * 
126*/127privateclass LogHandler implements Runnable
128    {
129130// 队列中没有数据时睡眠时间131privatefinalint WAITTIME = 1000;
132133        @Override
134publicvoid run()
135        {
136int count = 0;// 计数器137             Date writeTime = new Date();
138boolean time = false;
139             ConcurrentLinkedQueue<UserLog> uLogQueue = new ConcurrentLinkedQueue<UserLog>();
140while (true)
141            {
142try143                {
144                     UserLog log = uLogs.poll();
145if (log == null)
146                    {
147try148                        {
149// 睡眠1秒钟，防止队列中没有数据时线程在这里空跑150                            Thread.sleep(WAITTIME);
151                        }
152catch (Exception e)
153                        {}
154155// 如果时间到达了指定的阈值,将已经获取到的值提交156                         time = (System.currentTimeMillis() - writeTime
157                             .getTime()) > delayTime;
158if (time && (uLogQueue.size() > 0))
159                        {
160                             pool.execute(new UActionThread(uLogQueue));
161                             uLogQueue = new ConcurrentLinkedQueue<UserLog>();
162                             count = 0;
163                             writeTime = new Date();
164                        }
165continue;
166                    }
167168                    uLogQueue.add(log);
169                     count++;
170// 一次性处理的数据大小171if (count >= batchsize)
172                    {
173                         pool.execute(new UActionThread(uLogQueue));
174                         uLogQueue = new ConcurrentLinkedQueue<UserLog>();
175                         count = 0;
176                         writeTime = new Date();
177                    }
178                }
179catch (Throwable e)
180                {
181// 捕获所有异常,这个线程不能停止除非jvm停止了182                     log.error("UActionTools Log loop thread is exception.", e);
183                }
184            }
185        }
186    }
187 }