一个简单的消息处理框架

项目里碰到一个不错的消息处理框架，把核心代码抽出来做个备注。核心源码分几块。

1、一个消息管理类，在hashmap里创建了多个阻塞式消息队列

package com.fredric.demo;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class QueueManager {

    private Map map;

    public QueueManager(){

        map = new HashMap();

        //创建两个阻塞式队列
        map.put("queue_type1", new ArrayBlockingQueue(100));
        map.put("queue_type2", new ArrayBlockingQueue(100));
    }


    public boolean put(String queueType, Map data){
        BlockingQueue queue = (BlockingQueue) map.get(queueType);
        if(null == queue){
            throw new RuntimeException();
        }else{
            try {
                queue.put(data);
                return true;

            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return false;
        }
    }

    public Map take(String queueType){
        BlockingQueue queue = (BlockingQueue) map.get(queueType);
        if(null == queue){
            throw new RuntimeException();
        }else{
            try {
                Map map = (Map) queue.take();
                return map;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return null;
        }
    }
}

2、构筑消息的处理，即消息的消费者。这里要注意的是消息具体的处理业务是单独再起线程的，如下：

package com.fredric.demo;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Type1Consumer implements Runnable{

    private ExecutorService executorService;

    public Type1Consumer(){

        //创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程
        executorService = Executors.newCachedThreadPool();
    }


    public void run() {
        do{

            final Map msg = MsgProducer.queueManager.take("queue_type1");

            if(msg != null){
                executorService.execute(new Runnable() {
                    public void run() {
                        System.out.println("queue_type1 do msg handler" + msg.toString());
                    }
                });
            }

        }while(true);
    }
}

3、消息的生产者，即往消息队列管理类中插入数据

package com.fredric.demo;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;

public class MsgProducer {

    public static QueueManager queueManager = new QueueManager();


    public void initJobQueue(){
        //创建一个单线程化的线程池
        Executors.newSingleThreadExecutor().execute(new Type1Consumer());
        Executors.newSingleThreadExecutor().execute(new Type2Consumer());
    }


    public void sendType1Msg(String data){
        Map map = new HashMap<String, String>();
        map.put("time", new Date().toString());
        map.put("data", data);
        queueManager.put("queue_type1", map);
    }

    public void sendType2Msg(String data){
        Map map = new HashMap<String, String>();
        map.put("time", new Date().toString());
        map.put("data", data);
        queueManager.put("queue_type2", map);
    }
}

4、Main方法测试如下：

public class App {

    public static void main(String[] args){

        MsgProducer producer = new MsgProducer();
        producer.initJobQueue();

        for(int i = 0; i < 10; i++){
            producer.sendType1Msg("type1_msg");
            producer.sendType2Msg("type2_msg");
        }
    }
}