java-消息中间件-基于内存的mq

Posted 2021-02-13 leeethan

1. 如果用户的请求比较费时，可以考虑将用户的请求信息放到队列中，立即返回给用户处理中等信息，这样可以给用户比较流畅的体验，后端可以利用单独的服务消费消息，做到了解耦，提高了并发能力。

2. 本文使用jdk为我们提供的阻塞队列api，来实现一个基于内存的简单消息队列。主要涉及的接口BlockingQueue,以及它的实现类ArrayBlockingQueue（数组实现的）和LinkedBlockingQueue(链表实现的)。

3. BlockingQueue的主要方法
   添加元素
   put() //往队列里插入元素,如果队列已经满,则会一直等待直到队列为空插入新元素,或者线程被中断抛出异常;
offer() //往队列添加元素如果队列已满直接返回false,队列未满则直接插入并返回true;
add() //对offer()方法的简单封装.如果队列已满,抛出异常new IllegalStateException("Queue full");
   删除元素
   remove() //方法直接删除队头的元素;
take() //取出并删除队头的元素,当队列为空,则会一直等待直到队列有新元素可以取出,或者线程被中断抛出异常;
pool() //取出并删除队头的元素,当队列为空,返回null;
peek() //直接取出队头的元素,并不删除;
element() //对peek方法进行简单封装,如果队头元素存在则取出并不删除,如果不存在抛出异常NoSuchElementException();

4. 基于内存的队列，队列的大小依赖于JVM内存的大小，一般如果是内存占用不大且处理相对较为及时的都可以采用此种方法。如果你在队列处理的时候需要有失败重试机制，那么用此种队列就不是特别合适了，可以使用基于数据库的mq。

5. 使用示例，简单模仿基于生产者、消费者、消息队列模型的内存mq

*生产者模型

    public class Produce {
    
    private int id ;
    
    private BlockingQueue<String> queue;
    
    public Produce() {
    }
    public Produce(int id, BlockingQueue<String> queue) {
        this.id = id;
        this.queue = queue;
        System.out.println("创建了生产者"+ id + "号");
    }
    
    public void produce(String message){
        boolean add = this.queue.add(message);
        if(add){
            System.out.println("生产者"+id+"号，生产了一条消息，"+message);  
        }
    }
}

*消费者模型

public class Consumer {
    
    private int id;

    private BlockingQueue<String> queue;
    
    private static ScheduledExecutorService  executors = Executors.newScheduledThreadPool(1);
    
    public Consumer() {
    }
    
    public Consumer(BlockingQueue<String> queue, int id){
        this.id = id;
        this.queue = queue;
        System.out.println("创建了消费者：" + id +"号");
        consumer();
    }
    public void consumer(){
        
            executors.scheduleWithFixedDelay(new Runnable() {
                @Override
                public void run() {
                    System.out.println("消费者手里的线程池核武器收到了一个命令，此时队列中的任务数"+queue.size()+"个");
                    try {
                        String message = queue.take();
                        System.out.println("消费者： " + id + "号，开始消费了");
                        Thread.sleep(3000);
                        System.out.println("消费者： " + id + "消费结束了，"+message);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }                   
                }
            }, 0, 5, TimeUnit.SECONDS);
            
    }
}

*测试类

public class Test {

    public static void main(String[] args) {
        LinkedBlockingDeque<String> queue = new LinkedBlockingDeque<>(1000);
        Produce produce = new Produce(1, queue);
        new Consumer(queue, 1);
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        for(int i=0;i<20;i++){
            produce.produce("@"+i+"@");
        }   
    }
}

ps:消息队列无论在分布式还是在高并发概念中，都是一个非常重要的数据处理方式，本文只是我对消息中间件这个技术栈的第一次试水，后面会陆续增加基于数据库的mq，以及现在已经比较成熟商业中间件产品rabbitmq、rocketmq等的学习笔记。