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