大体思路
Redis本身提供了一个发布/订阅模式，但生产消费者模式需要我们自己去实现。
- 利用Redis中的队列，将新消息放入名称为
name的队列末尾，完成消息生产者。 - 启动一个线程，使用
brpop命令循环从name队列取第一个元素，获得消息，调用注册的消费者执行业务逻辑。
本文只是做了一个简单的思路演示，对消息重试、超时、多线程并未做控制和实现。
集成Redis
引入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
配置RedisTemplate
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(redisConnectionFactory);
// 设置Key使用String序列化
redisTemplate.setKeySerializer(new StringRedisSerializer());
return redisTemplate;
}
}
生产者
public class QueueSender {
private RedisTemplate<Object, Object> redisTemplate;
public QueueSender(RedisTemplate<Object, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}
public void sendMsg(String queue, Object msg) {
redisTemplate.opsForList().leftPush(queue, msg);
}
}
消费者
接口类
public interface MsgConsumer {
void onMessage(Object message);
void onError(Object msg, Exception e);
}
监听器
class QueueListener implements Runnable {
public static final Logger log = LoggerFactory.getLogger(QueueListener.class);
private RedisTemplate<Object, Object> redisTemplate;
private String queue;
private MsgConsumer consumer;
public QueueListener(RedisTemplate<Object, Object> redisTemplate, String queue, MsgConsumer consumer) {
this.redisTemplate = redisTemplate;
this.queue = queue;
this.consumer = consumer;
}
@Override
public void run() {
log.info("QueueListener start...queue:{}", queue);
while (RedisMqConsumerContainer.run) {
try {
Object msg = redisTemplate.opsForList().rightPop(queue, 30, TimeUnit.SECONDS);
if (msg != null) {
try {
consumer.onMessage(msg);
} catch (Exception e) {
consumer.onError(msg, e);
}
}
} catch (QueryTimeoutException ignored) {
} catch (Exception e) {
if (RedisMqConsumerContainer.run) {
log.error("Queue:{}", queue, e);
} else {
log.info("QueueListener exits...queue:{}", queue);
}
}
}
}
}
消费者容器
配置类
public class QueueConfiguration {
/**
* 队列名称
*/
private String queue;
/**
* 消费者
*/
private MsgConsumer consumer;
private QueueConfiguration() {
}
public static Builder builder() {
return new Builder();
}
String getQueue() {
return queue;
}
MsgConsumer getConsumer() {
return consumer;
}
public static class Builder {
private QueueConfiguration configuration = new QueueConfiguration();
public QueueConfiguration defaultConfiguration(MsgConsumer consumer) {
configuration.consumer = consumer;
configuration.queue = consumer.getClass().getSimpleName();
return configuration;
}
public Builder queue(String queue) {
configuration.queue = queue;
return this;
}
public Builder consumer(MsgConsumer consumer) {
configuration.consumer = consumer;
return this;
}
public QueueConfiguration build() {
if (configuration.queue == null || configuration.queue.length() == 0) {
if (configuration.consumer != null) {
configuration.queue = configuration.getClass().getSimpleName();
}
}
return configuration;
}
}
}
消费者容器
public class RedisMqConsumerContainer {
private static final Logger log = LoggerFactory.getLogger(RedisMqConsumerContainer.class);
private Map<String, QueueConfiguration> consumerMap = new HashMap<>();
private RedisTemplate<Object, Object> redisTemplate;
static boolean run;
private ExecutorService exec;
public RedisMqConsumerContainer(RedisTemplate<Object, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}
public void addConsumer(QueueConfiguration configuration) {
if (consumerMap.containsKey(configuration.getQueue())) {
log.warn("Key:{} this key already exists, and it will be replaced", configuration.getQueue());
}
if (configuration.getConsumer() == null) {
log.warn("Key:{} consumer cannot be null, this configuration will be skipped", configuration.getQueue());
}
consumerMap.put(configuration.getQueue(), configuration);
}
public void destroy() {
run = false;
this.exec.shutdown();
log.info("QueueListener exiting.");
while (!this.exec.isTerminated()) {
}
log.info("QueueListener exited.");
}
public void init() {
run = true;
this.exec = Executors.newCachedThreadPool(r -> {
final AtomicInteger threadNumber = new AtomicInteger(1);
return new Thread(r, "RedisMQListener-" + threadNumber.getAndIncrement());
});
consumerMap = Collections.unmodifiableMap(consumerMap);
consumerMap.forEach((k, v) -> exec.submit(new QueueListener(redisTemplate, v.getQueue(), v.getConsumer())));
}
}
测试
配置消费者
public class TestListener implements MsgConsumer {
private static Logger log = LoggerFactory.getLogger(TestListener.class);
@Override
public void onMessage(Object message) {
log.info("收到消息:" + message);
// 随机发生错误
int n = 1 / (((int) (Math.random() * 10)) / 2);
}
@Override
public void onError(Object msg, Exception e) {
log.error("发生错误,消息:{}", msg, e);
}
}
配置消费者容器
/**
* 配置redis消息队列消费者容器
*
* @param redisTemplate redis
* @return 消费者容器
*/
@Bean(initMethod = "init", destroyMethod = "destroy")
public RedisMqConsumerContainer redisMqConsumerContainer(@Autowired RedisTemplate<Object, Object> redisTemplate) {
RedisMqConsumerContainer config = new RedisMqConsumerContainer(redisTemplate);
config.addConsumer(QueueConfiguration.builder()
.queue("TEST")
.consumer(new TestListener())
.build());
return config;
}
/**
* 配置redis消息队列生产者
*
* @param redisTemplate redis
* @return 生产者
*/
@Bean
public QueueSender queueSender(@Autowired RedisTemplate<Object, Object> redisTemplate) {
return new QueueSender(redisTemplate);
}
测试类
@RunWith(SpringRunner.class)
@SpringBootTest
public class RedisMqApplicationTests {
@Autowired
private QueueSender queueSender;
@Test
public void contextLoads() {
for (int i = 0; i < 10; i++) {
queueSender.sendMsg("TEST","hello~~~~,序号:"+i);
}
}
}
输出结果
2019-07-22 22:42:07.914 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:0
2019-07-22 22:42:07.926 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:1
2019-07-22 22:42:07.934 ERROR 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 发生错误,消息:hello~~~~,序号:1
java.lang.ArithmeticException: / by zero
at com.suitwe.redismq.listener.TestListener.onMessage(TestListener.java:17) ~[classes/:na]
at com.suitwe.redismq.QueueListener.run(QueueListener.java:34) ~[classes/:na]
at java.util.concurrent.ExecutorsWorker.run(ThreadPoolExecutor.java:624) [na:1.8.0_202]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_202]
2019-07-22 22:42:07.940 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:2
2019-07-22 22:42:07.948 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:3
2019-07-22 22:42:07.954 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:4
2019-07-22 22:42:07.962 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:5
2019-07-22 22:42:07.971 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:6
2019-07-22 22:42:07.977 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:7
2019-07-22 22:42:07.977 ERROR 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 发生错误,消息:hello~~~~,序号:7
java.lang.ArithmeticException: / by zero
at com.suitwe.redismq.listener.TestListener.onMessage(TestListener.java:17) ~[classes/:na]
at com.suitwe.redismq.QueueListener.run(QueueListener.java:34) ~[classes/:na]
at java.util.concurrent.ExecutorsWorker.run(ThreadPoolExecutor.java:624) [na:1.8.0_202]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_202]
2019-07-22 22:42:07.984 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:8
2019-07-22 22:42:07.990 INFO 791 --- [disMQListener-1] c.suitwe.redismq.listener.TestListener : 收到消息:hello~~~~,序号:9
