每日学习笔记(19)

今天的工作需求如下：有一个生产者负责生成源数据，将数据插入到多个工作队列中，每个工作队列由一个线程池进行处理，即每个线程池中可以启动多个线程对与其对应的工作队列中的元素取出来进行处理。

我将业务需求抽象为一个“单生产者—多消费者集群”模型，写了一个模拟代码进行测试，明天打算把模型再套回到业务中去。

import java.text.DateFormat;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 测试工作队列和线程池
 * @author fayin.dyk
 *
 */
/**
 * 测试流程: 0,预先创建n个工作队列
 *             1,一个生产者线程不断生产QueueElement，将其放置到n个工作队列中
 *             2,每个工作队列有一个对应的线程池，其中的工作线程从队列中依次取
 *             QueueElement，并进行相应的处理
 */

public class TestWorkQueue {
    
    private static Map<String, WorkQueue> workQueueList = null;//n个商品库工作队列
    private static int QUEUE_NUM = 10;
    static {
        initWorkQueue();
    }
    
    /**
     * 初始化n个工作队列
     */
    private static void initWorkQueue() {
        //创建同样数量的工作队列
        workQueueList = new HashMap<String, WorkQueue>(QUEUE_NUM);//队列名称与对应的工作队列
        for (int i = 0; i < QUEUE_NUM; ++i) {
            String entityName = "auction" + i;
            WorkQueue workQueue = new WorkQueue();
            workQueueList.put(entityName, workQueue);
        }
    }
    /**
     * 单个的生产者线程,不断生产QueueElement，放置到n个工作队列中
     * @author fayin.dyk
     *
     */
    private static class ProducerThread implements Runnable {
        private boolean isNeedToStop = false;
        
        /**
         * 生成行键值
         * @param itemId 商品id
         * @return
         */
        private String generateRowKey(String itemId) {
            String result = "";
            DateFormat format = new java.text.SimpleDateFormat("yyyyMMddHHmmss");
            Date curDate = new Date();
            String curDateStr = format.format(curDate);
            result = curDateStr + "_" + itemId;
            return result;
        }
        
        private QueueElement generateQueueElement() {//随机生产element
            String msgType = ItemNotifyConstants.NM_item_add;//消息类型
            //商品id
            Random rnd = new Random();
            String rndItemId = String.valueOf(rnd.nextInt(10000));
            String rowKey = generateRowKey(rndItemId);
            String location = "hz";
            String title = "auction_" + rndItemId;
            QueueElement element = new QueueElement(msgType,rowKey, rndItemId,  rndItemId, location, title, rndItemId);
            return element;
        }
        /**
         * 分发到各个工作队列中
         */
        private void deployWorkElement(QueueElement element) {
            Collection<WorkQueue> queueSet = workQueueList.values();
            for (WorkQueue workQueue : queueSet) {
                workQueue.enQueue(element);
            }
        }
        public void run() {
            while (!isNeedToStop) {
                //生产QueueElement,放置到n个工作队列中
                QueueElement element = generateQueueElement();
                deployWorkElement(element);
                System.out.println("生产者生产商品id : " + element.getItemID() + "\n");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                }
            }
        }
    }
    
    /**
     * 消费者，每个消费者是与一个队列对应的线程池，其中的线程从此队列中取element，进行相应处理
     * @author fayin.dyk
     *
     */
    private static class Consumer {
        private final static int CORE_THREAD_NUM = 20;
        private final static int MAX_THREAD_NUM = 21;
        private final static int CAPACITY = 1000;
        public String queueName = "";//当前队列名称
        public WorkQueue workQueue = null;//与其对应的工作队列
        
        public Consumer(WorkQueue workQueue, String queueName) {
            this.workQueue = workQueue;
            this.queueName = queueName;
        }
        
        ThreadPoolExecutor indexerThreadPool = new ThreadPoolExecutor(
                CORE_THREAD_NUM, MAX_THREAD_NUM, 10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(CAPACITY));//处理对应队列的消费者线程
        
        public void process() {
            indexerThreadPool.execute(new IndexerJobThread(workQueue, queueName));
            indexerThreadPool.execute(new IndexerJobThread(workQueue, queueName));
        }
        
        private static class IndexerJobThread implements Runnable {
            public String queueName = "";//当前队列名称
            public WorkQueue workQueue = null;
            public IndexerJobThread(WorkQueue workQueue, String queueName) {
                this.workQueue = workQueue;
                this.queueName = queueName;
            }        
            public void run() {
                System.out.println("消费者,当前队列  " + queueName + "队列大小" + workQueue.queueSize() +  " 消费者线程id: " + Thread.currentThread().getId() + "\n");

                while (true) {
                    if (!workQueue.isEmpty()) {
                        QueueElement element = workQueue.deQueue();
                        if (element != null) {
                            System.out.println("消费者,当前队列  " + queueName + " 商品id: " + element.getItemID() + " " + "消费者线程id: " + Thread.currentThread().getId() + "\n");
                        }
                    }
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                    }
                }
            }
        }

    }
    /**
     * 为各个工作队列初始化对应的消费者(每个消费者包含一个线程池来进行处理)
     */
    private static void initConsumers() {
        Set<Entry<String, WorkQueue>> queueSet = workQueueList.entrySet();
        for (Entry<String, WorkQueue> workQueueEntry : queueSet) {
            String queueName = workQueueEntry.getKey();
            WorkQueue workQueue = workQueueEntry.getValue();
            System.out.print("queueName: " + queueName + "workQueue:" + workQueue + "\n");
            Consumer consumer = new Consumer(workQueue, queueName);
            consumer.process();
        }        
    }
    public static void main(String[] args) {
        //生产者启动
        Thread producer = new Thread(new ProducerThread());
        producer.start();
        //启动消费者群
        initConsumers();
        try {
            Thread.sleep(100000);
        } catch (InterruptedException e) {
        }
    }
}