How to subscribe redis key space event with spring

Jerry
April 8, 2024
157 views
0 votes
2 Answers

My Environment

Mac Ventura 13.6.3
Temurin 17
SpringBoot 3.2.1
org.springframework.boot:spring-boot-starter-data-redis
redis cluster running on local machine. (localhost: 7001, localhost: 7002, localhost: 7003

What I want to do

I want to receive expire event from redis cluster with spring.

What I did

I connected to cluster, and expired a key

CONFIG SET notify-keyspace-events Ex

SET hi 123
EXPIRE hi 3

My (java) code

Configuration

package kr.co.yogiyo.payo.infrastructure.temporal;

import io.lettuce.core.ReadFrom;
import kr.co.yogiyo.payo.infrastructure.temporal.service.RedisExpireEventService;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.data.redis.RedisConnectionDetails;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

@RequiredArgsConstructor
@Configuration
public class RedisConfig {

  private final RedisConnectionDetails redisConnectionDetails;

  @Bean
  public RedisConnectionFactory redisConnectionFactory() {
    RedisConnectionDetails.Cluster cluster = redisConnectionDetails.getCluster();
    RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration();
    for (RedisConnectionDetails.Node node : cluster.getNodes()) {
      clusterConfiguration.addClusterNode(new RedisNode(node.host(), node.port()));
    }

    LettuceClientConfiguration clientConfig =
        LettuceClientConfiguration.builder().readFrom(ReadFrom.REPLICA_PREFERRED).build();

    return new LettuceConnectionFactory(clusterConfiguration, clientConfig);
  }

  @Bean
  RedisMessageListenerContainer keyExpirationListenerContainer(
      RedisConnectionFactory connectionFactory, RedisExpireEventService redisExpireEventService) {
    RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
    listenerContainer.setConnectionFactory(connectionFactory);
    listenerContainer.addMessageListener(
        redisExpireEventService, new PatternTopic("__keyevent@*__:expired"));
    return listenerContainer;
  }
}

the service

package kr.co.yogiyo.payo.infrastructure.temporal.service;

import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class RedisExpireEventService implements MessageListener {
  @Override
  public void onMessage(Message message, byte[] pattern) {
    System.out.println("Message received: " + message.toString());
  }
}

application.yml

spring:
  main:
    banner-mode: off
    allow-bean-definition-overriding: true
  jackson:
    property-naming-strategy: SNAKE_CASE
  data:
    redis:
      host: ${PAYO_REDIS_MASTER_HOST:localhost}
      port: ${PAYO_REDIS_MASTER_PORT:7002}
      cluster:
        nodes: ${PAYO_REDIS_REPLICATION_NODES:localhost:7001,localhost:7002,localhost:7003}

spring runs well, but nothing happens when key expired.
(I set breakpoint and ran with debug mode, still nothing happens)

Working python code

I tried with python, it worked like charm… I want to subscribe the event with spring.

import redis


def main():
    # Connect to Redis
    r = redis.Redis(host='localhost', port=7002, db=0)

    # Subscribe to the '__keyevent@0__:expired' channel
    pubsub = r.pubsub()
    pubsub.psubscribe('__keyevent@*__:expired')

    # Start listening for messages
    for message in pubsub.listen():
        print("something expired! ", message)

if __name__ == "__main__":
    main()

something expired!  {'type': 'psubscribe', 'pattern': None, 'channel': b'__keyevent@*__:expired', 'data': 1}
something expired!  {'type': 'pmessage', 'pattern': b'__keyevent@*__:expired', 'channel': b'__keyevent@0__:expired', 'data': b'hi'}

could you guys let me know what am I missing in my java code?

Answers

Try this

@Configuration
public class RedisConfig {

    @Bean
    public RedisMessageListenerContainer keyExpirationListenerContainer(
            RedisConnectionFactory connectionFactory,
            RedisExpireEventService redisExpireEventService) {
        RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
        listenerContainer.setConnectionFactory(connectionFactory);
        listenerContainer.addMessageListener(
                redisExpireEventService,
                new PatternTopic("__keyevent@*__:expired"));
        return listenerContainer;
    }
}

@Service
public class RedisExpireEventService implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        // Handle the key expiration event
        String expiredKey = new String(message.getBody());
        System.out.println("Key expired: " + expiredKey);
    }
}

By default, the key expiry listener is disabled when initializing the application. You can enable it by adding the following to either your main application class or your RedisConfig class:

@EnableRedisRepositories(enableKeyspaceEvents = RedisKeyValueAdapter.EnableKeyspaceEvents.ON_STARTUP)

You can find more details about this in the Spring Data Redis Repositories documentation.

Here is a full working example:

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisKeyValueAdapter;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.repository.configuration.EnableRedisRepositories;

@SpringBootApplication
@EnableRedisRepositories(enableKeyspaceEvents = RedisKeyValueAdapter.EnableKeyspaceEvents.ON_STARTUP)
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    RedisMessageListenerContainer keyExpirationListenerContainer(RedisConnectionFactory factory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        MessageListener listener = (message, pattern) -> System.out.println("Key expired: " + message);
        container.setConnectionFactory(factory);
        container.addMessageListener(listener, new PatternTopic("__keyevent@*__:expired"));
        return container;
    }
}

And running the following on Redis:

# You don't need to run this as Spring Data Redis will
# set the notify-keyspace-events to Ex by default
# CONFIG SET notify-keyspace-events Ex

SET hi 123
EXPIRE hi 3

Prints the following after 3 seconds:

Key expired: hi

Please signup or login to give your own answer.

Click here to cancel reply.