基本介绍
核心原理：利用Redis的List列表实现，发布事件对应rpush，订阅事件对应lpop
问题一：Redis不是自带Pub/Sub吗？
redis自带的pub/sub有两个问题：
1.如果发布消息期间订阅方没有连到redis，那么这条消息就收不到了，即使重新连接上来也收不到
2.redis内部是用一个线程给所有订阅连接推数据的，V生产> V消费 的情况下还会主动断开连接，有性能隐患。感兴趣的可以多了解一下它的原理。
问题二：要实现怎样一个工具，或者说想要什么样的效果？
效果就是得到一个service对象，这个对象有以下两个重要功能：
1.有个publish方法可以调用，用来灵活地发布消息。想发布什么就发布什么，想给哪个topic发送就给哪个topic发送。
2.可以预定义一些订阅者，定义好当收到某个topic的消息后，该做什么处理。
编码内容
（一）接口定义
第一步要做的就是定义接口，一个是发布接口，我们需要这样一个接口来发布消息，消息内容可以是任何形式的对象
public interface MessagePublisher {
/**
* 发布消息
* @param topic 主题
* @param msg 消息内容
*/
void publish(String topic, Object msg);
}
第二个是订阅接口，我们需要依此实现观察者模式
public interface MessageConsumer {
/**
* 获取此消费者订阅的topic
* @return 订阅topic
*/
String getTopic();
/**
* 回调方法，收到消息后，此方法被触发
* @param topic topic
* @param msg 消息内容
*/
void onMessage(String topic, Object msg);
}
第三个就是转换接口，已知Redis不能直接存储Java对象，所以必须进行转换，这里我们选择用String形式进行存储。所以我们需要一个类型转换工具
public interface Translator {
/**
* 将对象序列化为字符串
* @param obj 对象
* @return 字符串
*/
String serialize(Object obj);
/**
* 将字符串反序列化为对象
* @param str 字符串
* @return 对象
*/
Object deserialize(String str);
}
（二）转换器实现——JsonTranslator
问题一：取出数据后如何转换成正确的对象？
在写入redis的时候同时也写入该对象的类型信息，然后取出的时候利用该类型信息进行转换即可。
public class JsonTranslator implements Translator {
private static ObjectMapper MAPPER = new ObjectMapper();
/**
* 缓存类信息，优化速度
*/
private Map<String, Class> classCache = new HashMap<>();
@Override
public String serialize(Object obj) {
Message message = new Message();
message.setClazz(obj.getClass().getName());
message.setData(encode(obj));
return encode(message);
}
@Override
public Object deserialize(String str) {
Message message = decode(str, Message.class);
String className = message.getClazz();
Class clazz = classCache.get(className);
if(clazz != null)
return decode(message.getData(), clazz);
try {
clazz = Class.forName(className);
classCache.put(className, clazz);
return decode(message.getData(), clazz);
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
private String encode(Object obj) {
try {
return MAPPER.writeValueAsString(obj);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
private <T> T decode(String str, Class<T> clazz) {
try {
return (T) MAPPER.readValue(str, Message.class);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Data
class Message { //保存类信息，是为了反序列化能够得到正确类型的对象
/**
* 类名（含路径）
*/
private String clazz;
/**
* 序列化后的对象
*/
private String data;
}
}
（三）核心实现——RedisPubSub
问题一：Redis配置如何处理？
我们将Redis的配置与这个MQ解耦，让用户配置连接池后再注入进来即可。
问题二：如何知道要监听哪些topic？
我们把容器中的Consumer实现类都注入进来，就可以通过getTopic方法得到总共需要监听哪些topic。
问题三：如何进行监听？
每个需要监听的topic开一个线程进行监听，监听方法就是循环调用blpop。
问题四：监听到消息后如何进行通知？
当得到topic的消息的时候，就回调订阅此topic的consumer的onMessage方法。
问题五：如何启动和关闭监听？
我们给MQ类提供两个方法start和stop。在注入容器的时候指明这两个分别是init和destroy方法，这样它就能随着容器启动和停止了。
public class RedisPubSub implements MessagePublisher{
//外部注入信息
private JedisPool jedisPool;
private List<MessageConsumer> consumerList;
/**
* 对象和字符串的转换器，默认使用JsonTranslator
*/
private Translator translator = new JsonTranslator();
//内部信息
/**
* key:topic
* value:此topic的订阅者
*/
private Map<String, List<MessageConsumer>> subcribeInfo;
private List<MessageListener> listeners;
public void setJedisPool(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
public void setConsumerList(List<MessageConsumer> consumerList) {
this.consumerList = consumerList;
subcribeInfo = new HashMap<>();
String topic;
List<MessageConsumer> topicConsumers;
//注入消费者后，整理好订阅情况
for(MessageConsumer consumer : consumerList) {
topic = consumer.getTopic();
topicConsumers = subcribeInfo.get(topic);
if(topicConsumers == null) {
topicConsumers = new ArrayList<>();
subcribeInfo.put(topic, topicConsumers);
}
topicConsumers.add(consumer);
}
}
public void setTranslator(Translator translator) {
this.translator = translator;
}
public void publish(String topic, Object msg) {
Jedis jedis = jedisPool.getResource();
jedis.rpush(topic,translator.serialize(msg));
jedis.close();
}
public void start() {
MessageListener listener;
//每个topic开一个监听线程进行监听
for(String topic : subcribeInfo.keySet()) {
listener = new MessageListener(topic, subcribeInfo.get(topic));
listener.start();
listeners.add(listener);
}
}
public void stop() {
//关闭所有监听器
for(MessageListener listener: listeners) {
listener.stop();
}
}
public class MessageListener implements Runnable {
/**
* 此监听器监听的topic
*/
private String topic;
/**
* 此topic的消费者
*/
private List<MessageConsumer> consumers;
/**
* 绑定线程
*/
private Thread t;
public MessageListener(String topic, List<MessageConsumer> consumers) {
this.topic = topic;
this.consumers = consumers;
}
/**
* 将数据反序列化
* @param msg 字符串消息
* @return 消息对象
*/
public Object deserialize(String msg) {
return translator.deserialize(msg);
}
public void run() {
String msg;
Object obj;
//从池中抓取一个连接用来监听redis队列
Jedis jedis = jedisPool.getResource();
while(!Thread.interrupted()) {
msg = jedis.blpop(1, topic).get(1);
obj = deserialize(msg);
//收到消息后告知所有消费者
for(MessageConsumer consumer:consumers) {
consumer.onMessage(topic, obj);
}
}
jedis.close(); //订阅结束后释放资源
}
public void start() {
t = new Thread(this);
t.start();
}
public void stop() { //利用中断打断线程的运行
t.interrupt();
}
}
}
使用案例
（一）定义好Consumer，注入为容器bean
@Component
public class TestConsumer implements MessageConsumer {
@Override
public void onMessage(String topic, Object message) {
System.out.println((SomeObject)message);
}
@Override
public String getTopic() {
return "test";
}
}
由于Ttranslator会将对象转换好，所以只要将Object强制转换成指定类型即可使用。
（二）全局配置
@Configuration
public class TestConfig {
@Bean
public JedisPool jedisPool() {
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(20);
jedisPoolConfig.setMaxIdle(5);
jedisPoolConfig.setMinIdle(1);
return new JedisPool(jedisPoolConfig, "127.0.0.1", 6379, 2000, "123456");
}
@Bean(value = "rediMQ", initMethod = "start", destroyMethod = "stop")
@Autowired
public RedisPubSub redisPubSub(List<MessageConsumer> consumers, JedisPool jedisPool) {
RedisPubSub redisPubSub = new RedisPubSub();
redisPubSub.setJedisPool(jedisPool);
redisPubSub.setConsumerList(consumers);
return redisPubSub;
}
}
@Autowired 配合方法参数的List<MessageConsumer> 就可以得到容器中所有的Consumer。
（三）引入使用
@Service
public class SomeService {
@Autowired
private MessagePublisher publisher;
public void someOperation() {
publisher.publish("test", new SomeObject());
}
}
只需要以MessagePublisher接口的身份引入就可以了。
原文地址：https://www.cnblogs.com/longfurcat/p/10920918.html
时间： 2024-11-13 08:03:41