redis的发布订阅

redis做为消息中间件,进行异步解耦

项目结构总览

redis配置

@Configuration
public class RedisConfig {
/**
* 创建消息监听容器
* @param factory
* @param messageListenerAdapter
* @return RedisMessageListenerContainer
* @date 2019年10月08日
* @throws
/
@Bean
public RedisMessageListenerContainer listenerContainer(RedisConnectionFactory factory, MessageListenerAdapter messageListenerAdapter) {
RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
redisMessageListenerContainer.setConnectionFactory(factory);
redisMessageListenerContainer.addMessageListener(messageListenerAdapter, new PatternTopic(“topic”));
return redisMessageListenerContainer;
}
/*
* MessageReceiver 消息的消费类
* @param receiver
* @return
/
@Bean
MessageListenerAdapter listenerAdapter(MessageReceiver receiver ) {
//给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”
//这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看
return new MessageListenerAdapter(receiver, “receiveMessage”);
}
/*
* 默认的工厂初始化redis操作模板
*/
@Bean
StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
return new StringRedisTemplate(connectionFactory);
}

消息发送者

// 这里采用定时发送消息

@EnableScheduling
@Component
public class MessageSender {
@Autowired
RedisTemplate redisTemplate;
@Scheduled(cron = “0 */1 * * * ?”)
public void sendMesToRedis() {
Person person = new Person();
person.setAge(1);person.setName(“测试1”);
String personStr = JSON.toJSON(person).toString();
redisTemplate.convertAndSend(“topic”,personStr);

}

}

消息接收者

@Component
public class MessageReceiver {
/**
* 消息的消费者,处理消息
* @param message
*/
public void receiveMessage(String message) {
System.out.println(“接收消息为:”);
System.out.println(message);
}

}

执行结果

定时器一分钟执行一次

与mq的比较

reids不支持AMQP，MQTT等协议，且不支持JMS规范，redis没有提供对这些协议的支持.redis消息不能进行持久化,容易丢失.适合于一些不重要的日志记录的项目,集成mq又太浪费.就比较适合用redis的发布订阅

底层代码分析

https://www.cnblogs.com/duanxz/p/6053520.html

消费消息的时候多了,几个不明字符,是因为解析的时候出了问题.可参考下面文章

https://www.cnblogs.com/donfaquir/p/10594199.html