多線程編程：多線程併發制單的開發記錄【一】

最近天氣很冷，記得保暖，我在帝都向各位問好。一次舒服的小長假回來，是否是都忘了增刪改查怎麼敲了，那就趕忙粘貼一下子吧，別等着公司把你優化掉。

好了，老習慣，首先給各位腦補一下多線程必備的知識。

進程和線程：

下圖是在來自知乎用戶的解釋，我的感受狠到位

       進程（Process）是計算機中的程序關於某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位，是操做系統結構的基礎。在早期面向進程設計的計算機結構中，進程是程序的基本執行實體；在當代面向線程設計的計算機結構中，進程是線程的容器。程序是指令、數據及其組織形式的描述，進程是程序的實體。

       線程，有時被稱爲輕量級進程(Lightweight Process，LWP），是程序執行流的最小單元。線程是程序中一個單一的順序控制流程。進程內一個相對獨立的、可調度的執行單元，是系統獨立調度和分派CPU的基本單位指運行中的程序的調度單位。在單個程序中同時運行多個線程完成不一樣的工做，稱爲多線程。

進程和線程的關係：

 

線程安全：

       何爲線程安全？就是應用中多個線程訪問某一個類（對象或方法）時，這個類始終能表現出正確的行爲，那麼這個類（對象或方法）就是線程安全的。

       線程安全就是多線程訪問時，採用了加鎖機制(在多線程安全上加鎖也是一門技術活，不是說對於共享資源你簡單加個同步關鍵字或定義成同步方法就OK了的，鎖的不合理則會大大影響程序的性能，甚至影響到業務，這裏本人親身經歷過一個問題，鎖粒度大小對程序的影響，參考：http://www.cnblogs.com/1315925303zxz/p/7561236.html)，當一個線程訪問該類的某個數據時，進行保護，其餘線程不能進行訪問直到該線程讀取完，其餘線程纔可以使用。不會出現數據不一致或者數據污染。

        線程不安全呢，就是不提供數據訪問保護，有可能出現多個線程前後更改數據形成所獲得的數據是髒數據。這裏的加鎖機制常見的如：synchronized。

由spring管理的線程池進行併發制單的業務設計：

業務需求：多個通道進行數據採集，數據採集就是拿着VIN碼去抓取數據，可是當VIN碼不少時，採集的速度就很慢，因此實施多線程多併發進行採集。

一、spring管理線程池的配置，這裏須要說明的一點就是核心線程數和最大線程數的配置必定要按照本身業務的併發量來設定，不然不只不會提高併發效率，反而會出現各類數據污染的狀況。

 1 　　 <!-- 異步線程池 -->
 2     <bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
 3         <!-- 核心線程數，最小線程數 默認爲1 -->
 4         <property name="corePoolSize" value="4" />
 5         <!-- 最大線程數，默認爲Integer.MAX_VALUE -->
 6         <property name="maxPoolSize" value="8" />
 7         <!-- 隊列最大長度，通常須要設置值>=notifyScheduledMainExecutor.maxNum；默認爲Integer.MAX_VALUE -->
 8         <property name="queueCapacity" value="2000" /> 
 9         <!-- 線程池維護線程所容許的空閒時間，默認爲60s -->
10         <property name="keepAliveSeconds" value="300" />
11         <!-- 線程池對拒絕任務（無線程可用）的處理策略，目前只支持AbortPolicy、CallerRunsPolicy；默認爲後者 -->
12         <property name="rejectedExecutionHandler">
13             <!-- AbortPolicy:直接拋出java.util.concurrent.RejectedExecutionException異常 -->
14             <!-- CallerRunsPolicy:主線程直接執行該任務，執行完以後嘗試添加下一個任務到線程池中，能夠有效下降向線程池內添加任務的速度 -->
15             <!-- DiscardOldestPolicy:拋棄舊的任務、暫不支持；會致使被丟棄的任務沒法再次被執行 -->
16             <!-- DiscardPolicy:拋棄當前任務、暫不支持；會致使被丟棄的任務沒法再次被執行 -->
17             <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
18         </property>
19     </bean>

　　　　　<!-- 一汽大衆和奧迪多併發制單 -->
　　　　  <bean id="multiThreadTask_VW_AD" class="cn.base.thread.MultiThreadTask_VW_AD"/>

 

二、線程制單業務(這裏是僞代碼)，將符合條件的VIN碼進行數據採集，不然記錄到指定容器中統一處理(退單)。

初始化任務(VIN碼)隊列：

 1 public 
 2 class InitQueue {
 3     
 4     private static final Logger logger = LoggerFactory.getLogger(MultiThreadTask_VW_AD.class);
 5     
 6     /**
 7      *    併發隊列
 8      */
 9     private static Queue<String> queue = null;
10     
11     //初始化併發隊列
12     private static void initQueue(){
13         logger.info("初始化併發隊列：{}");
14         if(queue == null){
15             queue = new ConcurrentLinkedQueue<String>();    //併發隊列    
16         }
17         String tasklist = "JF1GH78F18G03614,JF1SH95F6AG110830,JF1SJ94D7DG010387333,JF1SH92F9CG26924,JF1SH92F5BG215090,JF1SH92F5BG222556,JF1SH92F4CG279994,JF1BR96D7CG114298,JF1BR96D0BG078632,JF1SH95F9AG094011,JF1SH98FXAG186997,JF1BM92D8BG022510,JF1BM92DXAG013855,JF1BM94D8EG0366";
18         String[] split = tasklist.split(",");
19         List<String> task = Arrays.asList(split);    //數組轉集合
20         queue.addAll(task);        //按照集合中元素的順序將集合中所有元素放進隊列
21     }
22     
23     //對外提供一個任務隊列
24     public static Queue<String> getQueue(){
25         initQueue();
26         return queue;
27     }
28 
29 }

View Code

制單：

 1 public 
   class MultiThreadTask_VW_AD implements Runnable,Serializable{
 4     private static final Logger logger = LoggerFactory.getLogger(MultiThreadTask_VW_AD.class);
 5     private static final long serialVersionUID = 0;
 6     
 7     private static final Object lock = new Object();    //靜態鎖，任務隊列在某一時刻只能由一個線程操做
 8     private static List<String> errorVinList = new ArrayList<>();    //存放不合理VIN碼的集合
 9     
10     private static Queue<String> taskQueue = InitQueue.getQueue();    //任務隊列
11 
12     //數據採集
13     @Override
14     public void run() {
15         synchronized (lock) {
16             int queueSize = taskQueue.size();
17             if(queueSize > 0){
18                 logger.info("任務隊列目前有{}個任務",queueSize);
19                 String thisVin = taskQueue.poll();
20                 logger.info("當前任務VIN碼：{}",thisVin);
21                 if(thisVin.length() == 17){
22                     System.out.println("採集任務已完成【"+Thread.currentThread().getName()+"】");
23                 }else{
24                     System.out.println("不進行採集【"+Thread.currentThread().getName()+"】");
25                     errorVinList.add(thisVin);
26                     System.err.println("未採集的VIN碼有："+errorVinList);
27                 }
28             }else{
29                 logger.info("任務隊列目前無任務");
30             }
31         }
32     }
33     
34 }

       再次強調：加鎖的時候必定要掌握好鎖的粒度，否則會影響併發效率。

三、模擬多通道進行數據採集。

 1 public 
 2 class ThreadPoolController {
 3     
 4     private static final Logger logger = LoggerFactory.getLogger(ThreadPoolController.class);
 5     
 6     /**
 7      * 一汽大衆、奧迪多併發進行數據採集：
 8      */
 9     @Test
10     public void vwadTest(){
11         //加載線程池配置文件
12         ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:springs/applicationContext-threadpool.xml");
13         //經過applicationContext獲取異步線程池對象threadPoolTaskExecutor
14         ThreadPoolTaskExecutor threadPool = (ThreadPoolTaskExecutor) ctx.getBean("threadPoolTaskExecutor");
15         //獲取品牌制單業務對象(由spring管理的)
16         MultiThreadTask_VW_AD multiThreadTask_VW_AD = (MultiThreadTask_VW_AD) ctx.getBean("multiThreadTask_VW_AD");
17         for(int i=1; i<=15; i++){
18             threadPool.execute(multiThreadTask_VW_AD);
19         }
20         
21     }

 

後期繼續更新多線程併發的文章，由於被這傢伙給看上了，不得不深刻剖析了，一直在學習的路上。