多线程——worker-master模式

落日映苍穹つ 2022-06-09 02:12 139阅读 0赞

该模式通过将任务拆分，多个worker线程共同处理任务，最终返回结果给master，由master组装最终结果。

一、master

需要保存的数据

1、任务队列，需要注意线程安全，保存需要处理的数据

2、worker线程队列

3、子任务处理结果集，需要注意线程安全

构造器

需要传入worker和线程数量，建立worker线程，存入worker线程队列中

需要的方法如下

1、分解任务存入任务队列

2、启动所有的worker线程

3、返回任务任务的子结果集

4、判断所有worker线程是否执行完

二、worker

需要保存的数据

1、任务队列，保存需要处理的数据，和master共用

2、子任务处理结果集，和master共用

需要的方法如下

1、从master中获得任务队列

2、从master中获得初始的空结果集

3、返回结果集

4、任务处理逻辑

5、线程逻辑，从任务队列中出队尚未处理的数据，如果没有数据，则结束线程，否则处理任务，将结果添加到结果集

以下给一个多线程进行两个集合比较的例子。线程根据cpu核数给，不要给太大。当集合在1000的时候多线程开始有优势，10000的话提升效率非常明显

master

package com.wlf.demo.masterWorker.master;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Queue;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    import com.wlf.demo.masterWorker.worker.Worker;
    
    public class Master {
    
    	// 任务队列
    	// 其中为需要处理的数据
    	protected Queue<Object> workQueue = new ConcurrentLinkedQueue<Object>();
    	
    	// worker进程队列
    	protected Map<String, Thread> threadMap = new HashMap<String, Thread>();   
    	
    	//子任务处理结果集  
        protected Map<String, Object> resultMap = new ConcurrentHashMap<String, Object>();  
        
        public Master(Worker worker, int countWorker) {  
            worker.setWorkQueue(workQueue);  
            worker.setResultMap(resultMap);  
            for(int i=0; i<countWorker; i++) {  
                threadMap.put(Integer.toString(i),   
                        new Thread(worker, Integer.toString(i)));  
            }  
        }  
        
        //是否所有的子任务都执行完毕
        public boolean isComplete() {  
            for(Map.Entry<String, Thread> entry : threadMap.entrySet()) {  
                if(entry.getValue().getState() != Thread.State.TERMINATED)  
                    //存在为完成的线程  
                    return false;  
            }  
            return true;  
        }  
        
        //提交一个任务  
        public void submit(Object data) {  
            workQueue.add(data);  
        }  
        
     	//返回任务结果集  
        public Map<String, Object> getResultMap() {  
            return resultMap;  
        }  
        
        //执行所有Worker进程，进行处理  
        public void execute() {  
            for(Map.Entry<String, Thread> entry : threadMap.entrySet()) {  
                entry.getValue().start();  
            }  
        }  
        
    }

worker

package com.wlf.demo.masterWorker.worker;
    
    import java.util.Map;
    import java.util.Queue;
    
    public abstract class Worker implements Runnable {
    
    	// 任务队列
    	protected Queue<Object> workQueue;
    	
    	//子任务处理结果集  
        protected Map<String, Object> resultMap;
    	
    	@Override
    	public void run() {
    		while(true) {
    			// 获取需要处理的数据
    			Object data = workQueue.poll();
    			if(data == null) 
    				break;  
    			//处理子任务  
                Object re = handle(data);  
                //将处理结果写入结果集  
                resultMap.put(Integer.toString(data.hashCode()), re);  
    		}
    	}
    	
    	//子任务处理的逻辑，在这里不作具体实现，由子类实现  
        public abstract Object handle(Object data); 
        
    	public Queue<Object> getWorkQueue() {
    		return workQueue;
    	}
    
    	public void setWorkQueue(Queue<Object> workQueue) {
    		this.workQueue = workQueue;
    	}
    
    	public Map<String, Object> getResultMap() {
    		return resultMap;
    	}
    
    	public void setResultMap(Map<String, Object> resultMap) {
    		this.resultMap = resultMap;
    	}
    
    }

package com.wlf.demo.masterWorker.worker.impl;
    
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    import com.wlf.demo.masterWorker.worker.Worker;
    
    public class CompareListWorker extends Worker{
    
    	private static List<Integer> listCompareB;
    	
    	public CompareListWorker(List<Integer> list) {
    		listCompareB = Collections.synchronizedList(list);
    	}
    	
    	@SuppressWarnings("unchecked")
    	@Override
    	public Object handle(Object data) {
    		System.out.println(Thread.currentThread().getName()+":work");
    		long time = System.currentTimeMillis();
    		Map<String,List<Integer>> result = new HashMap<String,List<Integer>>();
    		List<Integer> listCompareA = new ArrayList();
    		listCompareA.addAll((List<Integer>) data);
    		
    		// 交集
    		List<Integer> listRetain = new ArrayList<Integer>();
    		listRetain.addAll(listCompareA);
    		boolean isRetainAll = listRetain.retainAll(listCompareB);
    		
    		// listCompareB中多余的元素
    		listCompareB.removeAll(listCompareA);
    		result.put("listB-more",listCompareB);
    		
    		// 差集
    		boolean isDifference  = listCompareA.removeAll(listRetain);
    		// listCompareA中的元素全部在listCompareB中
    		if(!isRetainAll) {
    			result.put("retain", listRetain);
    			result.put("listA-more", new ArrayList<Integer>());
    		}
    		else {
    			result.put("retain", listRetain);
    			// listCompareB中没有的元素
    			result.put("listA-more", listCompareA);
    		}
    		
    		time = System.currentTimeMillis()-time;
            System.out.println(time+"ms");
    		return result;
    	}
    	
    	public static List<Integer> getList() {
    		return listCompareB;
    	}
    
    }

多线程调用

package com.wlf.demo.masterWorker;
    
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.Collections;
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    
    import com.wlf.demo.masterWorker.master.Master;
    import com.wlf.demo.masterWorker.worker.impl.CompareListWorker;
    
    public class ApplicationCompareList {
    
    	public static final int LIST_SIZE=10000;
    	public static final int WORK_SIZE=2;
    	
    	public static void main(String[] args) { 
    		
    		/*
    		List<Integer> listAllAFix = Arrays.asList(
    				new Integer[]{
    						1,2,3,4,5,6,7,8,9,10,
    						11,12,13,14,15,16,17,18,19,20,
    						21,22,23,24,25,26,27,28,29,30,31,32,33
    		});
    		
    		List<Integer> listAllBFix = Arrays.asList(
    				new Integer[]{
    						0,1,2,3,4,5,6,7,8,9,10,
    						11,12,13,14,15,16,17,18,19,20,
    						21,22,23,24,25,26,27,28,29,30
    		});
    		
    		
    		List<Integer> listAllA = new ArrayList<Integer>(listAllAFix);
    		List<Integer> listAllB = new ArrayList<Integer>(listAllBFix);
    		*/
    		
    		List<Integer> listAllA = new LinkedList<Integer>();
    		List<Integer> listAllB = new LinkedList<Integer>();
    		
    		for(int i=0;i<10000;i++) {
    			listAllA.add(i);
    		}
    		
    		listAllB.addAll(listAllA.subList(2000, listAllA.size()));
    		
    		for(int i=0;i<2000;i++) {
    			listAllB.add(10000+i);
    		}
    		
    		Master master = new Master(new CompareListWorker(listAllB), WORK_SIZE);
    		
    		int size = listAllA.size();
    		int loopCount = size%LIST_SIZE == 0 ?  size/LIST_SIZE : size/LIST_SIZE+1;
    		
    		for(int i=0; i<loopCount; i++){ 
                if(i==loopCount-1){
                	master.submit(listAllA.subList(i*LIST_SIZE, size));  
                }else{
                	master.submit(listAllA.subList(i*LIST_SIZE, (i+1)*LIST_SIZE));  
                }
    		}
    		
    		long time = System.currentTimeMillis();
            master.execute(); //开始计算  
    		
            Map<String, Object> resultMap = master.getResultMap();  
            
            List<Integer> listRetain = new ArrayList<Integer>();  //完全匹配的list
            List<Integer> listAMore = new ArrayList<Integer>(); //listA多出来的元素
            List<Integer> listBMore = new ArrayList<Integer>(); //listB多出来的元素
            
            //需要等待所有Worker都执行完即可  
            while(true) {  
                Set<String> keys = resultMap.keySet();  //开始计算最终结果  
                String key = null;  
                for(String k : keys) {  
                    key = k;  
                    break;  
                }  
                Map<String,List<Integer>> result = null;  
                if(key != null) {
                	result = (Map<String,List<Integer>>) resultMap.get(key);
                    resultMap.remove(key); //移除已被计算过的项目  
                }
                if(result != null) {
                	if(result.get("retain")!=null)
                		listRetain.addAll(result.get("retain"));
                	if(result.get("listA-more")!=null)
                		listAMore.addAll(result.get("listA-more"));
                }      
                if(master.isComplete() && resultMap.size()==0) {
                	listBMore.addAll(CompareListWorker.getList());
                    break;  
                }
            }  
            
    //        System.out.println("-----------------retain-----------------"); 
    //        for(Integer item : listRetain) {
    //        	 System.out.println(item);
    //        }
    //        
    //        System.out.println("-----------------listA-more-----------------"); 
    //        for(Integer item : listAMore) {
    //        	 System.out.println(item);
    //        }
    //        
    //        System.out.println("-----------------listB-more-----------------"); 
    //        for(Integer item : listBMore) {
    //        	 System.out.println(item);
    //        }
      
            time = System.currentTimeMillis()-time;
            System.out.println(time+"ms");
            
    	}
    	
    }

单线程调用

package com.wlf.demo.masterWorker;
    
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.HashMap;
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    
    import com.wlf.demo.masterWorker.master.Master;
    import com.wlf.demo.masterWorker.worker.impl.CompareListWorker;
    import com.wlf.demo.masterWorker.worker.impl.PlusWorker;
    
    public class ApplicationSingleCompareList {
    
    	public static void main(String[] args) { 
    		
    		long time = System.currentTimeMillis();
    		
    /*		
    		List<Integer> listAllAFix = Arrays.asList(
    				new Integer[]{
    						1,2,3,4,5,6,7,8,9,10,
    						11,12,13,14,15,16,17,18,19,20,
    						21,22,23,24,25,26,27,28,29,30,31,32,33
    		});
    		
    		List<Integer> listAllBFix = Arrays.asList(
    				new Integer[]{
    						0,1,2,3,4,5,6,7,8,9,10,
    						11,12,13,14,15,16,17,18,19,20,
    						21,22,23,24,25,26,27,28,29,30
    		});
    		
    		
    		List<Integer> listAllA = new ArrayList<Integer>(listAllAFix);
    		List<Integer> listAllB = new ArrayList<Integer>(listAllBFix);*/
    		
    		
    		List<Integer> listAllA = new LinkedList<Integer>();
    		List<Integer> listAllB = new LinkedList<Integer>();
    		
    		for(int i=0;i<10000;i++) {
    			listAllA.add(i);
    		}
    		
    		listAllB.addAll(listAllA.subList(2000, listAllA.size()));
    		
    		for(int i=0;i<2000;i++) {
    			listAllB.add(10000+i);
    		}
    		
    		// 交集
    		List<Integer> listRetain = new ArrayList<Integer>();
    		listRetain.addAll(listAllA);
    		boolean isRetainAll = listRetain.retainAll(listAllB);
    		
    		// listCompareB中多余的元素
    		listAllB.removeAll(listAllA);
    		
    		// 差集
    		listAllA.removeAll(listRetain);
    		
    		Map<String,List<Integer>> result = new HashMap<String,List<Integer>>();
    		// listCompareA中的元素全部在listCompareB中
    		if(!isRetainAll) {
    			result.put("retain", listRetain);
    			result.put("listA-more", new ArrayList<Integer>());
    		}
    		else {
    			result.put("retain", listRetain);
    			// listCompareB中没有的元素
    			result.put("listA-more", listAllA);
    		}
    		result.put("listB-more", listAllB);
    		
            List<Integer> listAMore = new ArrayList<Integer>(); //listA多出来的元素
            List<Integer> listBMore = new ArrayList<Integer>(); //listB多出来的元素
    		
    		if(result != null) {
            	if(result.get("listA-more")!=null)
            		listAMore.addAll(result.get("listA-more"));
            }      
    		listBMore.addAll(listAllB);
    		
    //		System.out.println("-----------------retain-----------------"); 
    //		for(Integer item : listRetain) {
    //			System.out.println(item);
    //		}
    //      
    //		System.out.println("-----------------listA-more-----------------"); 
    //		for(Integer item : listAMore) {
    //			System.out.println(item);
    //		}
    //      
    //		System.out.println("-----------------listB-more-----------------"); 
    //		for(Integer item : listBMore) {
    //			System.out.println(item);
    //		}
    		
            time = System.currentTimeMillis()-time;
            System.out.println(time+"ms");
      
    	}
    	
    }