前言

常见的消息中间件MQ有很多，如Apache的ActiveMQ，阿里的RocketMQ，使用Erlang编写的RabbitMQ，以及我使用的Redis作MQ
Redis作为NoSQL数据库，经常被当做缓存使用，好处就不用多说了
如果把Redis作为消息中间件实现MQ，分别使用512Bytes、1K和10K四个不同大小的测试数据进行10万次写入和10万次读取操作。如果每次写入的数据比较小的情况下性能是要高于RabbitMQ，如果单次写入数据大小超过10K，Redis表现会很慢；而出队时无论数据大小，RabbitMQ和Redis表现不相上下。
综上，Redis完全可以作为轻量级MQ来使用。

我的整个服务是基于Spring实现的

目录结构如下

目录结构.png

这里我写了一个生产者队列和一个消费者队列

public class QueueProducerRunnable implements Runnable{
    private QueueProducer queueProducer;
    private long sleepTime;

    public QueueProducerRunnable(QueueProducer queueProducer, long sleepTime) {
        this.queueProducer = queueProducer;
        this.sleepTime = sleepTime;
    }

    @Override
    public void run(){
        do {
            try {
                queueProducer.lpushData();
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }while (true);
    }
}

创建消费者

public class QueueConsumerRunnable implements Runnable{
    private QueueConsumer queueConsumer;
    long sleepTime;

    public QueueConsumerRunnable(QueueConsumer queueConsumer, long sleepTime) {
        this.queueConsumer = queueConsumer;
        this.sleepTime = sleepTime;
    }

    @Override
    public void run(){
        do {
            try {
                queueConsumer.rpopData();
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }while (true);
    }
}

生产者生产队列

import com.lyl.test.common.RedisUtil;
import org.springframework.stereotype.Component;

@Component("QueueProducer")
public class QueueProducer {
    private static String queuePrefix = "Queue.Data";

    public void lpushData(){
        /*--入队--*/
        /*
        下面只是存入redis一些可以看到顺序的数据
        开发中去实现自己的逻辑
         */
        long now = System.currentTimeMillis();
        String time = String.valueOf(now);
        String str = time.substring(8,10);

        //lpush操作把数据插入到表头
        RedisUtil.lpush(5,queuePrefix,str);

    }
}

消费者消费队列

import com.lyl.test.common.RedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

@Component("QueueConsumer")
public class QueueConsumer {
    private static String queuePrefix = "Queue.Data";

    public void rpopData(){
        String str = RedisUtil.rpop(5,queuePrefix);

        if (StringUtils.isEmpty(str)){
            return;
        }

        System.out.println(str);
    }
}

生产者和消费者现在已经创建完毕，我需要在服务启动的时候创建线程进行生产和消费，这里我监听了Spring内置的ContextStartedEvent事件，我需要在root application context初始化完成后创建线程，root application context 没有parent，可以通过这个判断Spring容器的状态

import com.lyl.test.common.RedisUtil;
import com.lyl.test.queue.QueueConsumer;
import com.lyl.test.queue.QueueProducer;
import com.lyl.test.queue.runnable.QueueConsumerRunnable;
import com.lyl.test.queue.runnable.QueueProducerRunnable;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStream;
import java.util.PropertyResourceBundle;

@Component("QueueListener")
public class QueueListener implements ApplicationListener<ContextRefreshedEvent> {
    //生产者个数
    private static int PRODUCER_QUEUE_THREAD;
    //生产者线程等待时间
    private static long PRODUCER_QUEUE_SLEEP;
    //消费者个数
    private static int CONSUMER_QUEUE_THREAD;
    //消费者线程等待时间
    private static long CONSUMER_QUEUE_SLEEP;

    @Resource
    private QueueProducer queueProducer;

    @Resource
    private QueueConsumer queueConsumer;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        if(contextRefreshedEvent.getApplicationContext().getParent() == null){
            System.out.println("消息队列启动成功");
            //启动生产者队列
            for (int i=0;i<PRODUCER_QUEUE_THREAD;i++){
                QueueProducerRunnable queueProducerRunnable = new QueueProducerRunnable(queueProducer,PRODUCER_QUEUE_SLEEP);
                Thread thread = new Thread(queueProducerRunnable);
                thread.start();
            }

            //启动消费者队列
            for (int i=0;i<CONSUMER_QUEUE_THREAD;i++){
                QueueConsumerRunnable queueConsumerRunnable = new QueueConsumerRunnable(queueConsumer,CONSUMER_QUEUE_SLEEP);
                Thread thread = new Thread(queueConsumerRunnable);
                thread.start();
            }
        }
    }

    static {
        try {
        //初始化配置
        InputStream fis = RedisUtil.class.getClassLoader().getResourceAsStream("queue.properties");
        PropertyResourceBundle props = new PropertyResourceBundle(fis);

        PRODUCER_QUEUE_THREAD = Integer.valueOf(props.getString("producer_queue_thread"));
        PRODUCER_QUEUE_SLEEP = Long.valueOf(props.getString("producer_queue_sleep"));

        CONSUMER_QUEUE_THREAD = Integer.valueOf(props.getString("consumer_queue_thread"));
        CONSUMER_QUEUE_SLEEP = Long.valueOf(props.getString("consumer_queue_sleep"));

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

对Redis的操作实际上是使用出栈入栈的思想，根据不同的业务场景采用先进先出，或先进后出。我这里使用的是先进先出策略，使用lpush命令把需要处理的数据加入队列头部，通过rpop命令从队尾取出元素。如果要实现先进后出，在队尾插入元素同时在队尾取出元素。

适用的业务场景

基于Redis的轻量级队列服务适用于同步要求较低，而要实现的操作耗时较长，可以放入消息队列中进行处理。比如商品订单的回调通知就可以用这种模式去实现，回调通知必须保证要发送成功一次，但是第一次发送不一定能成功，这就需要去储存发送失败的订单进行后续的发送任务。

参考网站

Redis命令参考