本文以redis为数据结构基础，配合Spring管理机制，使用java实现了一个轻量级、可配置的消息队列。适合的项目特点：

• Spring框架管理对象


• 有消息需求，但不想维护mq中间件


• 有使用redis


• 对消息持久化并没有很苛刻的要求

需要使用rabbitmq实现延迟消息请参考这里

1. 设计方案

设计主要包含以下几点：

• 将整个Redis当做消息池，以kv形式存储消息


• 使用ZSET做优先队列，按照score维持优先级


• 使用LIST结构，以先进先出的方式消费


• zset和list存储消息地址（对应消息池的每个key）


• 自定义路由对象，存储zset和list名称，以点对点的方式将消息从zset路由到正确的list


• 使用定时器维持路由



• 根据TTL规则实现消息延迟

  

2. 代码实现

2.1 技术说明

示例使用Springboot，gradle，redis，jdk8。

2.2 核心代码

核心代码主要包含消息对象Message，路由器Route和消息队列RedisMQ。

2.2.1 Message对象

package git.yampery.msmq;
/**
* @decription Message
*

封装消息元数据

2.2.2 Route(消息路由器)

package git.yampery.msmq;
/**
* @decription Route
*

消息路由器，主要控制将消息从指定的队列路由到待消费的list
* 通过这种方式实现自定义延迟以及优先级发送

2.2.3 RedisMq(消息队列)

package git.yampery.msmq;
import git.yampery.utils.JedisUtils;
import org.springframework.scheduling.annotation.Scheduled;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
/**
* @decription RedisMQ
*

基于redis的消息队列

2.2.4 RedisMq消息队列配置：

• mq.properties文件

# 队列的监听数量
mq.monitor.count =30
# 队列一
mq.queue.first =queue:1
# 队列二
mq.queue.second =queue:2
# 消费列表一
mq.consumer.first =list:1
# 消费列表二
mq.consumer.second =list:2

• MqConfig.java文件

package git.yampery.config;
import git.yampery.msmq.RedisMQ;
import git.yampery.msmq.Route;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import java.util.ArrayList;
import java.util.List;
/**
* @decription MqConfig
*

消息队列配置






* ----------------------------------------------------
*/
@Configuration
public class MqConfig {
@Bean(name = "redisMQ")
@Primary
public RedisMQ getRedisMq() {
RedisMQ redisMQ = new RedisMQ();
// 配置监听队列元素数量
redisMQ.setMonitorCount(monitorCount);
// 配置路由表
redisMQ.setRoutes(routeList());
return redisMQ;
}
/**
* 返回路由表
* @return
*/
public List routeList() {
List routeList = new ArrayList<>();
Route routeFirst = new Route(queueFirst, listFirst);
Route routeSecOnd= new Route(queueSecond, listSecond);
routeList.add(routeFirst);
routeList.add(routeSecond);
return routeList;
}
@Value("${mq.monitor.count}")
private int monitorCount;
@Value("${mq.queue.first}")
private String queueFirst;
@Value("${mq.queue.second}")
private String queueSecond;
@Value("${mq.consumer.first}")
private String listFirst;
@Value("${mq.consumer.second}")
private String listSecond;
}


如果使用的是xml配置　请参考：

<bean id="redisMQ" class="git.yampery.msmq.RedisMQ">
<property name="monitorCount" value="15"/>
<property name="routes">
<list>
<bean class="git.yampery.msmq.Route">
<property name="queue" value="${mq.queue.first}"/>
<property name="list" value="${mq.consumer.first}"/>
bean>
<bean class="git.yampery.msmq.Route">
<property name="queue" value="${mq.queue.second}"/>
<property name="list" value="${mq.consumer.second}"/>
bean>
list>
property>
bean>

2.2.5 消费者

并没有内置消费者监听器来实现，可以直接使用定时器实现

package git.yampery.task;
import com.alibaba.fastjson.JSONObject;
import git.yampery.msmq.RedisMQ;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
import java.util.List;
/**
* @decription MsgTask
*

发送消息

4. 测试

测试设置20秒延迟，发布消息到queue:1，在list:1消费。

package git.yampery.mq;
import com.alibaba.fastjson.JSONObject;
import git.yampery.msmq.Message;
import git.yampery.msmq.RedisMQ;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
import java.util.UUID;
/**
* @decription TestMQ
*

测试


* @author Yampery
* @date 2018/2/9 18:43
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class TestMQ {
@Resource
private RedisMQ redisMQ;
@Value("${mq.queue.first}")
private String MQ_QUEUE_FIRST;
@Test
public void testMq() {
JSONObject jObj = new JSONObject();
jObj.put("msg", "这是一条短信");
String seqId = UUID.randomUUID().toString();
// 将有效信息放入消息队列和消息池中
Message message = new Message();
message.setBody(jObj.toJSONString());
// 可以添加延迟配置
message.setDelay(20);
message.setTopic("SMS");
message.setCreateTime(System.currentTimeMillis());
message.setId(seqId);
// 设置消息池ttl，防止长期占用
message.setTtl(20 * 60);
message.setStatus(0);
message.setPriority(0);
redisMQ.addMsgPool(message);
redisMQ.enMessage(MQ_QUEUE_FIRST,
message.getCreateTime() + message.getDelay() + message.getPriority(), message.getId());
}
}


3. 总结

文章利用redis已有的数据存储结构，实现了轻量级的消息队列，并未真正实现消息持久化。示例是针对点对点的消息路由方式，当然，也可以扩展成广播和主题的方式，不过，这样就得不偿失了，如果需求比较复杂，可靠性要求较高，反而不如直接维护rabbitmq之类的消息队列。
需要使用rabbitmq实现延迟消息请参考这里

4. 源码连接

文章并未贴出所有代码，gradle构建、jedisUtils以及一些配置等可以参考源码，源码只需要设置自己的redis配置即可。
https://github.com/Yampery/rdsmq.git