spring-redis-mq
基于Spring和Redis的分布式消息队列（MessageQueue）
使用方法
创建项目
由于这个库还没有提交到Maven的中央仓库，所以需要手动将其导入到你的私人仓库中。首先fork源码到本地后使用mvn package打包。
然后添加到本地仓库：
mvn install:install-file
-DgroupId=com.scienjus
-DartifactId=spring-redis-mq
-Dversion=1.0-SNAPSHOT
-Dpackaging=jar
-Dfile=/path/to/jar/spring-redis-mq.jar
所有依赖 Jar：
<properties>
<spring.version>4.1.8.RELEASE</spring.version>
<jedis.version>2.7.3</jedis.version>
<aspectj.version>1.8.7</aspectj.version>
<quartz.version>2.2.1</quartz.version>
</properties>
<dependencies>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>${jedis.version}</version>
</dependency>
<!-- For quartz -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>${quartz.version}</version>
</dependency>
<!--For aop-->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<version>${aspectj.version}</version>
</dependency>
</dependencies>
配置Spring Bean
配置Jedis客户端：
@Bean
public JedisPool jedisPool() {
return new JedisPool("127.0.0.1", 6379);
}
配置消费者：
@Bean
public Consumer consumer() {
RedisConsumer consumer = new RedisConsumer();
consumer.setJedisPool(jedisPool());
return consumer;
}
配置生产者：
@Bean
public Producer producer() {
RedisProducer producer = new RedisProducer();
producer.setJedisPool(jedisPool());
return producer;
}
配置消费者定时扫描任务（仅当使用注解驱动的消费者时才需要配置）：
@Bean(initMethod = "init")
public SchedulerBeanFactory schedulerBeanFactory() {
SchedulerBeanFactory schedulerBeanFactory = new SchedulerBeanFactory();
schedulerBeanFactory.setConsumer(consumer());
return schedulerBeanFactory;
}
注意一定要将initMethod设为init方法。
配置生产者自动推送任务（仅当使用注解驱动的生产者时才需要配置）：
@Bean
public ProducerWorker producerWorker() {
ProducerWorker producerWorker = new ProducerWorker();
producerWorker.setProducer(producer());
return producerWorker;
}
创建生产者实例
方法1：注入producer，调用sendMessage方法：
@Component
public class SayHelloProducer {
@Autowired
private Producer producer;
public void sayHello(String name) {
producer.sendMessage("say_hello", new Message(name));
}
}
方法2：使用@ToQueue注解，retrun需要发送的对象（需要配置producerWorker）：
@Producer
public class UserServiceImpl implements UserService {
@Autowired
private UserDao userDao;
@ToQueue(topic = "new_user") //添加新的用户后，将其发送到消息队列
public User insert(User user) {
this.userDao.insert(user);
return user;
}
}
创建消费者实例
方法1：注入consumer，调用getMessage方法（需要自己开线程循环获取）：
@Component
public class SayHelloConsumer {
@Autowired
private Consumer consumer;
public void sayHello() {
Message message;
while ((message = consumer.getMessage("say_hello")) != null) {
System.out.println("Hello ! " + message.getContent() + " !");
}
}
}
方法2：为类添加@Consumer注解，为对应的方法添加@OnMessage注解（需要配置schedulerBeanFactory）：
@Consumer
public class SayHelloConsumer {
@OnMessage(topic = "say_hello")
public void onSayHello(String name) {
System.out.println("Hello ! " + name + " !");
}
}
消费者的重试机制
当@OnMessage方法的返回值类型为boolean类型，并且执行的结果为false时，系统认定此消息执行失败。
消息执行失败后，系统会将这个消息重新插入到消息队列中（顺序排在最后）。
通过@ToQueue的expire属性可以设置消息的生存时间（单位为秒），默认为永不过期。
当消息的生存时间超过后还没有消费成功，系统将会丢掉这个消息。
一个简单的例子：
//生产者
@Producer
public class UserServiceImpl implements UserService {
@Autowired
private UserDao userDao;
@ToQueue(topic = "new_user", expire = 24 * 3600) //添加新的用户后，将其发送到消息队列，消息的生存时间是24小时
public User insert(User user) {
this.userDao.insert(user);
return user;
}
}
@Consumer
public class NewUserConsumer {
@OnMessage(value = "new_user") //如果邮件发送失败，需要尝试重新发送。
public boolean onNewUser(User user) {
try {
//发送邮件
MailSender.sendWelcomeMail(user.getEmail(), user.getNickname());
//发送成功，任务完成，返回true
retrun true;
} catch (Exception e) {
//发送失败，尝试重试，返回false
retrun false;
}
}
}
当然，如果一个消费方法永远不会失败（或是失败后不需要重试），可以直接设置为void方法。
PS：之前版本使用重试次数控制失败处理。但是系统修复需要的是时间，而重试次数很有可能会被浪费掉，因此改为了生存时间。
待办事项
- 消息处理失败的处理（重新插入队列，设置消息的生存周期）
- 将队列分为两个阶段，等待投递和等待接收
- 监控页面
帮助
我的邮箱：[email protected]
由于 Redis 本身的限制，这个项目并不适合使用在生产环境中，在此推荐 Redis 作者开发的消息队列 Disque。一些介绍：
- 该项目的地址：Disque, an in-memory, distributed job queue
- 该项目的中文介绍：Disque 使用教程
- Java 的客户端实现（Jedis 的作者开发）：Jedisque
- 与这个项目用法相同的Disque实现：Spring Dsique