spring-boot-starter-data-redis

本文精选：https://www.yuque.com/mrcode.cn/note-combat/wxot719u4r6iaoxt

spring-boot-starter-data-redis 的使用在 官方文档 - 3.0.6 (opens new window) 中提到，而它依赖的是 spring-data-redis:3.0.6 (opens new window)

发布订阅模式使用

需要注意的是：发布订阅模式，需要订阅的客户端在线，才能收到消息，离线期间的消息是收不到的

发布消息

package cn.mrcode.handler;

import com.alibaba.fastjson2.JSONObject;
import cn.mrcode.CollectionTaskCreateEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * @author mrcode
 * @date 2023/11/15 18:54
 */
@Component
public class CollectionTaskEventHandler {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @EventListener
    @Async
    public void taskPublish(CollectionTaskCreateEvent event) {
        CollectionTaskMsg message = new CollectionTaskMsg();
        message.setTime(LocalDateTime.now());
        message.setId(event.getId());
         // 调用 convertAndSend 发送消息
        // 消息内容需要自己转换为 字符串
        stringRedisTemplate.convertAndSend(CollectionTaskSubcribe.TOPIC, JSONObject.toJSONString(message));
    }
}

订阅消息

订阅消息麻烦一点

手动方式

自己获取 connection 然后订阅

 stringRedisTemplate.getConnectionFactory()
	.getConnection()
	.subscribe(listener,"topic 名称".getBytes());

spirng 容器方式

暴露 RedisMessageListenerContainer 对象

package cn.mrcode.redis;

import cn.hutool.core.collection.CollectionUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.util.List;

/**
 * @author mrcode
 * @date 2023/11/15 19:03
 */
@Configuration
public class MyRedisConfig {
    @Autowired(required = false)
    private List<MyMessageListener> listeners;

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        if (CollectionUtil.isNotEmpty(listeners)) {
            for (MyMessageListener listener : listeners) {
                container.addMessageListener(listener, new ChannelTopic(listener.getTopicName()));
            }
        }
        return container;
    }
}

MyMessageListener 是我们自己定义的一个接口

package cn.mrcode.redis;

import org.springframework.data.redis.connection.MessageListener;

public interface MyMessageListener extends MessageListener {
    /**
     * 该消息订阅器要监听的 topic
     * @return
     */
    String getTopicName();
}

然后实现 MyMessageListener 接口

package cn.mrcode.handler;

import cn.mrcode.redis.MyMessageListener;
import org.springframework.data.redis.connection.Message;
import org.springframework.stereotype.Component;

/**
 * @author mrcode
 * @date 2023/11/15 19:18
 */
@Component
public class CollectionTaskSubcribe implements MyMessageListener {
    public static final String TOPIC = "CollectionTask";

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String json = new String(message.getBody());
        // 前面发送的时候，转成的是 json 字符串，这里拿到之后，再用 json 工具反序列化为消息对象就行
        System.out.println(json);
    }

    @Override
    public String getTopicName() {
        return TOPIC;
    }
}