java后端项目拓展redis监听器_java redis监听

java后端项目拓展redis监听器

1 检查redis配置文件

2 修改application.yml文件配置redis；创建redids配置类，代码如下

@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.port}")
    private int port;

    @Value("${spring.redis.timeout}")
    private int timeout;

    @Value("${spring.pool.max-idle}")
    private int maxIdle;

    @Value("${spring.pool.max-wait}")
    private long maxWaitMillis;

    @Value("${spring.redis.password}")
    private String password;


    @Bean
    public JedisPoolConfig getJedisPoolConfig() {
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxIdle(maxIdle);
        jedisPoolConfig.setMaxWaitMillis(maxWaitMillis);
        return jedisPoolConfig;
    }

    @Bean
    public JedisConnectionFactory jedisConnectionFactory() {

        if (host.split(",").length == 1) {
            //单机版
            logger.info("-----------------blaze redis 单机版-------------------------");
            JedisConnectionFactory factory = new JedisConnectionFactory();
            factory.setHostName(host.split(":")[0]);
            //factory.setPort(Integer.valueOf(host.split(":")[1]));
            factory.setPort(port);
            factory.setPassword(password);
            factory.setTimeout(timeout);
            factory.setPoolConfig(getJedisPoolConfig());
            return factory;
        } else {
            //集群
            logger.info("-----------------blaze redis 集群版-------------------------");
            JedisConnectionFactory jcf = new JedisConnectionFactory(getClusterConfiguration());
            jcf.setPoolConfig(getJedisPoolConfig());
            jcf.setPassword(password); //集群的密码认证
            return jcf;
        }
    }

   /* @Bean
    public CacheManager cacheManager(RedisTemplate redisTemplate) {
        logger.info("===cacheManager successed");
        RedisCacheManager redisCacheManager = new RedisCacheManager(redisTemplate);
        //设置缓存过期时间
        redisCacheManager.setDefaultExpiration(1800);//秒
        return redisCacheManager;
    }*/

    @Bean(name = "redisTemplate")
    public RedisTemplate<String, Object> getRedisTemplate() {
        logger.info("===redisTemplate successed");
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new Jackson2JsonRedisSerializer(Object.class));
        redisTemplate.setConnectionFactory(jedisConnectionFactory());
        return redisTemplate;
    }


    @Bean
    public RedisClusterConfiguration getClusterConfiguration() {
        if (host.split(",").length > 1) {
            //如果是host是集群模式的才进行以下操作
            Map<String, Object> source = new HashMap<String, Object>();
            source.put("spring.redis.cluster.nodes", host);
            source.put("spring.redis.cluster.timeout", timeout);
            //source.put("spring.redis.cluster.max-redirects", redirects);
            source.put("spring.redis.cluster.password", password);
            return new RedisClusterConfiguration(new
                    MapPropertySource("RedisClusterConfiguration", source));
        } else {
            return null;
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

3到这里redis基础配置已完成，接下来实现key过期监听和分布式分布式锁的创建

redis监听器-配置类

4监听器配类代码如下

@Configuration
public class RedisListenerConfig {
    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 处理乱码
     * @return
     */
    @Bean
    public RedisTemplate redisTemplateInit() {

        // key序列化
        redisTemplate.setKeySerializer(new StringRedisSerializer());

        //val实例化
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

        return redisTemplate;
    }

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

5这里提供单机版和自定义监听接口和实现方式，大家凭借业务自行选择

代码依次如下：

1自定义接口代码

/**
 * ClassName: RedisListener <br/>
 * Description:  自定义redis监听器接口<br/>
 * date: 2024-02-27 14:58<br/>
 *
 * @author lj<br />
 * @since JDK 11
 */
public interface RedisListener {

    void message(String key);
}
1
2
3
4
5
6
7
8
9
10
11
12

2单机版监听实现类代码

package com.example.test.listener;

import com.example.test.constant.RedisPrefix;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * ClassName: RedisKeyExpirationListener <br/>
 * Description: 单机版redis 监听器<br/>
 * date: 2024-02-27 11:20<br/>
 *
 * @author lj<br />
 * @since JDK 11
 */

@Slf4j
@Component
public class RedisKeyExpirationListener  extends KeyExpirationEventMessageListener {

    @Autowired
    private RedisListener redisListener;
    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }


    /**
     * 针对redis数据失效事件，进行数据处理
     * @param message 失效的key
     */
  /*  @Override
    public void onMessage(Message message, byte[] pattern) {
        log.info("过期redis数据：" + message.toString());
        try {
            String key = message.toString();
            //从失效key中筛选代表订单失效的key
            String orderWithKey = RedisPrefix.ORDER_BLAZE;
            if (key.startsWith(orderWithKey) && !StringUtils.isEmpty(key)){
                log.info("订单号为【】超时未支付-自动修改为已取消状态");
                System.out.println("ok!");
            }else {
                System.out.println("NO!");
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【修改支付订单过期状态异常】：" + e.getMessage());
        }
    }*/
    @Override
    public void onMessage(Message message, byte[] pattern) {
        log.info("单机版redis监听到key:" + message.toString());
        redisListener.message(message.toString());
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

3创建所需工厂bean

@Service
public class RedisKeyExpirationBeanFactory {

    //初始化所有的RedisKeyExpirationService对象到map中(key为beanId，value为bean对象)
    @Autowired
    private Map<String, RedisKeyExpirationService> beans = new ConcurrentHashMap<>(2);

    /**
     * 根据type类型获取不同的实例对象
     */
    public RedisKeyExpirationService getInstance(String type) {
        RedisKeyExpirationService service = beans.get(type);
        if (service == null) {
            throw new RuntimeException("过期key类型暂不支持");
        }
        return service;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

4监听redis过期key统一入口

/**
 * TODO //、、监听redis过期key统一入口
 * ClassName: RedisKeyExpirationService <br/>
 * Description: <br/>
 * date: 2024-02-27 15:04<br/>
 *
 * @author lj<br />
 * @since JDK 11
 */
public interface RedisKeyExpirationService {

    /**
     * @param expiredKey 过期key
     * @return void
     * @Description TODO    统一执行方法
     * @date 2020/10/15 15:20
     * @author Fuqiang
     */
    void execute(String expiredKey);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

5 redis-Key前缀常量池

/**
 * ClassName: RedisPrefix <br/>
 * Description: redis-Key前缀常量池<br/>
 * date: 2024-02-27 14:44<br/>
 *
 * @author lj<br />
 * @since JDK 11
 */
public class RedisPrefix {


    /**
     * 订单超时自动过期前缀
     */
    public static final String ORDER_BLAZE = "order_blaze:";

    public static final String COURSE_COUPON_GET_TIME = "course_coupon_get_time:";

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

6分布式锁

/**
 * ClassName: RedisLock <br/>
 * Description: 分布式锁 <br/>
 * date: 2024-02-27 15:08<br/>
 *
 * @author lj<br />
 * @since JDK 11
 */
public class RedisLock {

    private StringRedisTemplate stringRedisTemplate;


    public RedisLock(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    /**
     * @param key     操作key
     * @param value   value
     * @param exptime 过期时间
     * @return boolean          是否成功boolean值，成功返回true，失败返回false
     * @Description setNx       加锁重载
     * @date 2020/10/15 10:49
     * @author lj
     */
    public boolean lock(String key, Object value, long exptime) {
        boolean isLock = setIfAbsent(key, value, exptime);
        return isLock;
    }

    /**
     * @param key   redis里的key
     * @param value redis里的value
     * @return boolean  成功状态，成功=true，失败=false
     * @Description 删除key，解锁
     * @date 2020/10/15 10:54
     * @author lj
     */
    public boolean unlock(String key, Object value) {
        /** lua */
        final String RELEASE_LOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 最后释放锁
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(RELEASE_LOCK_LUA_SCRIPT, Long.class);
        Long result = stringRedisTemplate.execute(redisScript, Collections.singletonList(key), value);
        return result == 1;

    }


    /**
     * @param key     redis的key
     * @param value   redis的value
     * @param exptime 过期时间
     * @return boolean 是否加锁成功
     */
    public boolean setIfAbsent(String key, Object value, long exptime) {
        Boolean result = (Boolean) stringRedisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
                RedisSerializer valueSerializer = stringRedisTemplate.getValueSerializer();
                RedisSerializer keySerializer = stringRedisTemplate.getKeySerializer();
                Object obj = connection.execute("set", keySerializer.serialize(key),
                        valueSerializer.serialize(value),
                        "NX".getBytes(StandardCharsets.UTF_8),
                        "EX".getBytes(StandardCharsets.UTF_8),
                        String.valueOf(exptime).getBytes(StandardCharsets.UTF_8));
                return obj != null;
            }
        });
        return result;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

7自定义监听器实现结合redis_lock分布式锁

/**
 * ClassName: RedisListenerImpl <br/>
 * Description: 自定义redis监听器实现<br/>
 * date: 2024-02-27 14:59<br/>
 *
 * @author lj<br />
 * @since JDK 11
 */
@Component
@Log4j2
public class RedisListenerImpl  implements RedisListener{

    private String[] redisListenerKey = {
            RedisPrefix.ORDER_BLAZE,RedisPrefix.COURSE_COUPON_GET_TIME
    };

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedisKeyExpirationBeanFactory beanFactory;

    @Override
    public void message(String expiredKey) {
        log.info("========RedisListenerImpl=========");
        log.info("redis监听过期key，前缀：{}", expiredKey);
        for (String key : redisListenerKey) {
            if (expiredKey.startsWith(key)) {
                //截取第一个:之前的字符串
                String substring = expiredKey.substring(0, expiredKey.indexOf(":"));
                RedisLock lockRedis = new RedisLock(stringRedisTemplate);
                boolean lock = lockRedis.lock(expiredKey, "_lock_flag", 10L);
                if (lock) {
                    try {
                        log.info("redis分布式锁加锁成功，锁标识：{}", expiredKey);
                        beanFactory.getInstance(substring).execute(expiredKey);
                    } finally {
                        boolean unlock = lockRedis.unlock(expiredKey, "_lock_flag");
                        if (unlock) {
                            log.info("redis分布式锁解锁成功，锁标识：{}", expiredKey);
                        }
                    }
                }
                break;
            }
        }
    }


}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

结合业务调用接口实现过期监听测试

效果如下：