SpringBoot之Redis定时发送消息

41次阅读

共计 7721 个字符，预计需要花费 20 分钟才能阅读完成。

实时发送定时公告，倒计时功能通过监听 Redis 缓存过期（Key 失效）事件。类似用途可以用于订单定时关闭，商品或活动上下架。

############################# EVENT NOTIFICATION ##############################

# Redis can notify Pub/Sub clients about events happening in the key space.
# This feature is documented at http://redis.io/topics/notifications
#
# For instance if keyspace events notification is enabled, and a client
# performs a DEL operation on key "foo" stored in the Database 0, two
# messages will be published via Pub/Sub:
#
# PUBLISH __keyspace@0__:foo del
# PUBLISH __keyevent@0__:del foo
#
# It is possible to select the events that Redis will notify among a set
# of classes. Every class is identified by a single character:
#
#  K     Keyspace events, published with __keyspace@<db>__ prefix.
#  E     Keyevent events, published with __keyevent@<db>__ prefix.
#  g     Generic commands (non-type specific) like DEL, EXPIRE, RENAME, ...
#  $     String commands
#  l     List commands
#  s     Set commands
#  h     Hash commands
#  z     Sorted set commands
#  x     Expired events (events generated every time a key expires)
#  e     Evicted events (events generated when a key is evicted for maxmemory)
#  A     Alias for g$lshzxe, so that the "AKE" string means all the events.
#
#  The "notify-keyspace-events" takes as argument a string that is composed
#  of zero or multiple characters. The empty string means that notifications
#  are disabled.
#
#  Example: to enable list and generic events, from the point of view of the
#           event name, use:
#
#  notify-keyspace-events Elg
#
#  Example 2: to get the stream of the expired keys subscribing to channel
#             name __keyevent@0__:expired use:
#
#  notify-keyspace-events Ex
#
#  By default all notifications are disabled because most users don't need
#  this feature and the feature has some overhead. Note that if you don't
#  specify at least one of K or E, no events will be delivered.
notify-keyspace-events ""

__keyevent@__:expired 其实是 指所有库，可以指定库下标监听 16 个默认数据库的某一个，比如 keyevent@1 指定壹号 库。打开redisclientA，PSUBSCRIBE 指令订阅事件。

127.0.0.1:6379> PSUBSCRIBE __keyevent@*__:expired
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "__keyevent@*__:expired"
3) (integer) 1

再开启另一个redisclientB，发送过期数据指定事件 2 秒。

127.0.0.1:6379> setex test 2 2
OK

redisclientA 就会监听到 redisclientB 过期 key

127.0.0.1:6379> PSUBSCRIBE __keyevent@*__:expired
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "__keyevent@*__:expired"
3) (integer) 1
1) "pmessage"
2) "__keyevent@*__:expired"
3) "__keyevent@0__:expired"
4) "test"

KeyExpirationEvent1MessageListener 可参考 org.springframework.data.redis.listener.KeyExpirationEventMessageListener 源码实现的（默认订阅的是keyevent@*:expired），而我们目标是监听壹号库。因为 0 号库给限流和 oauth2 用了，里面存在很多短期 key，会监听许多不相干的业务 key 缓存。此外，不能给 KeyExpirationEvent1MessageListener 加上 @Component，因为存在 bean 循环依赖问题，可以通过 SpringContextHolder 解决。

@Slf4j
public class KeyExpirationEvent1MessageListener extends KeyExpirationEventMessageListener {private static final Topic KEYEVENT1_EXPIRED_TOPIC = new PatternTopic("__keyevent@1__:expired");

    /**
     * @param listenerContainer must not be {@literal null}.
     */
    public KeyExpirationEvent1MessageListener(RedisMessageListenerContainer listenerContainer) {super(listenerContainer);
    }

    @Override
    public void doRegister(RedisMessageListenerContainer listenerContainer) {listenerContainer.addMessageListener(this, KEYEVENT1_EXPIRED_TOPIC);
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 获取过期的 key
        String expireKey = message.toString();
        // 设置监听频道
        if (expireKey.startsWith(RedisConstant.NOTIFY_RECEIVE)) {log.info("过期的键值对的消息 ID:" + expireKey);
            log.info("消息监听频道 topic:" + new String(message.getChannel()));
// 获取消息发送 id，通过
            String sendId = expireKey.substring(RedisConstant.NOTIFY_RECEIVE.length());
            SysNotifySendService sysNotifySendService =  SpringContextHolder.getBean(SysNotifySendService.class);
//common 服务提供 websocket 发送远程接口 RemoteCommonService
            RemoteCommonService remoteCommonService =  SpringContextHolder.getBean(RemoteCommonService.class);
            SysNotifyReceiveService receiveService =  SpringContextHolder.getBean(SysNotifyReceiveService.class);
            SysNotifySend sysNotifySend = sysNotifySendService.getOne(Wrappers.<SysNotifySend>lambdaQuery().eq(SysNotifySend::getSendId, sendId));
            com.gdjs.gold.admin.api.dto.Message websocketMsg = new com.gdjs.gold.admin.api.dto.Message();
            websocketMsg.setSendId(sysNotifySend.getSendId());
            websocketMsg.setFrom(sysNotifySend.getSendUserId());
            websocketMsg.setDestination(ThemeEnum.BUSINESS.getTheme());
            websocketMsg.setMessage(sysNotifySend.getContent());
            remoteCommonService.sendMessage(websocketMsg);     
        }
    }

}

@Configuration
public class RedisListenerConfig {

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, ApplicationContext context) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
// 通过 new 监听器，并且往 RedisMessageListenerContainer 注册监听器
        KeyExpirationEvent1MessageListener listener = new KeyExpirationEvent1MessageListener(container);
        listener.doRegister(container);
        listener.setApplicationEventPublisher(context);
        return container;
    }

}

JedisConnectionFactory jedisConnectionFactory = (JedisConnectionFactory) stringRedisTemplate.getConnectionFactory();
jedisConnectionFactory.setDatabase(切换到指定的 db 上);
stringRedisTemplate.setConnectionFactory(jedisConnectionFactory);

SpringBoot 2.X 之后的版本，RedisConnectionFactory 动态切换库必须是 LettuceConnectionFactory，必须是配置 RedisTemplate 时候指定，下面看源码。

public class RedisUtil {
    /**
     * 切换 redis 数据库
     *
     * @param redisTemplate springboot 封装的 redis 对象
     * @param index         数据库下标
     */
    public static void select(RedisTemplate redisTemplate, int index) {LettuceConnectionFactory lettuceConnectionFactory = (LettuceConnectionFactory) redisTemplate.getConnectionFactory();
        if (lettuceConnectionFactory != null) {lettuceConnectionFactory.setDatabase(index);
            redisTemplate.setConnectionFactory(lettuceConnectionFactory);
            lettuceConnectionFactory.resetConnection();}
    }

}

/**
 * @author caochikai
 * @date 2019/7/12
 * Redis 配置类
 */
@EnableCaching
@Configuration
@AllArgsConstructor
@AutoConfigureBefore(RedisAutoConfiguration.class)
public class RedisTemplateConfig {
//@AllArgsConstructor 是构造器注入进来 LettuceConnectionFactory
    private final LettuceConnectionFactory lcfactory;

    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        // 关闭共享链接，动态切换的重点在这里
        **lcfactory.setShareNativeConnection(false);**
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
        redisTemplate.setConnectionFactory(lcfactory);
        return redisTemplate;
    }

    @Bean
    public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForHash();
    }

    @Bean
    public ValueOperations<String, String> valueOperations(RedisTemplate<String, String> redisTemplate) {return redisTemplate.opsForValue();
    }

    @Bean
    public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForList();
    }

    @Bean
    public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForSet();
    }

    @Bean
    public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {return redisTemplate.opsForZSet();
    }
}

key 需要指定 key 命名规则前缀 RedisConstant.NOTIFY_RECEIVE 常量（通常取表名），随后加上表的主键。时间计算间距小技巧，博主业务使用的是 LocalDateTime，Duration.between(LocalDateTime.now(), 指定发送时间).getSeconds()，过期时间单位使用的是 TimeUnit 枚举。切换到指定一号库后，记得切换回来零号库，减少对其他业务的影响

/**
     * redis 缓存过期监听
     *
     * @param sysNotifySend 定时发送消息
     * @param message 消息内容
     */
    private void redisSetMsgKey(SysNotifySend sysNotifySend, Message message) {LocalDateTime sendTime = sysNotifySend.getSendTime();
        String jsonString = JSONUtil.parseObj(message).toJSONString(0);
        RedisUtil.select(redisTemplate, 1);
        redisTemplate.opsForValue().set(RedisConstant.NOTIFY_RECEIVE + sysNotifySend.getSendId(), jsonString, Duration.between(LocalDateTime.now(), sendTime).getSeconds(), TimeUnit.SECONDS);
        RedisUtil.select(redisTemplate, 0);
    }

SpringBoot2.0 以上整合 redis 根据下标动态切换数据库
redis 缓存过期策略，监听 redis 缓存

正文完

发表至： springboot

2020-05-25

0

关于springboot:SpringBoot-简介

关于springboot:记录本周遇到的问题

关于springboot:SpringBoot-在线协同办公小程序开发-全栈式项目实战

关于springboot:Spring-Boot优雅整合Swagger2自动生成在线文档

Python-中-staticmethod-和-classmethod-原理探究

MySQL-的-crashsafe-原理解析