devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

异步编程实战之lettuce

作者：你好赵伟 | 2024-03-07 01:27:53

赞

踩

异步编程实战之lettuce

一, 添加所需依赖

  1. <parent>
  2.   <groupId>org.springframework.boot</groupId>
  3.   <artifactId>spring-boot-starter-parent</artifactId>
  4.   <version>2.6.5</version>
  5.   <relativePath/> <!-- lookup parent from repository -->
  6. </parent>
  7. <dependency>
  8.   <groupId>org.springframework.boot</groupId>
  9.   <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
  10. </dependency>
  11. <dependency>
  12.   <groupId>org.apache.commons</groupId>
  13.   <artifactId>commons-pool2</artifactId>
  14. </dependency>

二, redis配置参数

  1. # redis
  2. spring.redis.database=0
  3. spring.redis.host=r-uf63gkx4qs0ygz3rag.redis.rds.aliyuncs.com
  4. spring.redis.port=6379
  5. spring.redis.password=ywjAYKDz28RFCJg9iRW4
  6. # 连接超时时间
  7. spring.redis.timeout=5000
  8. # 连接池最大连接数（使用负值表示没有限制）
  9. spring.redis.lettuce.pool.max-active=5000
  10. # 连接池中的最小空闲连接
  11. spring.redis.lettuce.pool.min-idle=1
  12. # 连接池中的最大空闲连接
  13. spring.redis.lettuce.pool.max-idle=100
  14. # 连接池最大阻塞等待时间（使用负值表示没有限制）
  15. spring.redis.lettuce.pool.max-wait=1000
  16. #在关闭客户端连接之前等待任务处理完成的最长时间，在这之后，无论任务是否执行完成，都会被执行器关闭，默认100ms
  17. spring.redis.lettuce.shutdown-timeout=1000
  18. #是否缓存空值
  19. spring.cache.redis.cache-null-values=false

三, redis连接池设置

  1. package com.shuinfo.ods_poc.config;
  2.  
  3. import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
  4. import org.springframework.beans.factory.annotation.Value;
  5. import org.springframework.cache.annotation.EnableCaching;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  8. import org.springframework.data.redis.connection.RedisPassword;
  9. import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
  10. import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
  11. import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
  12. import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
  13. import org.springframework.data.redis.core.ReactiveRedisTemplate;
  14. import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
  15. import org.springframework.data.redis.serializer.RedisSerializationContext;
  16. import org.springframework.data.redis.serializer.RedisSerializer;
  17. import org.springframework.data.redis.serializer.StringRedisSerializer;
  18. import java.time.Duration;
  19.  
  20. @Configuration
  21. @EnableCaching
  22. public class RedisConfig {
  23.     @Value("${spring.redis.database}")
  24.     private int database;
  25.  
  26.     @Value("${spring.redis.host}")
  27.     private String host;
  28.  
  29.     @Value("${spring.redis.password}")
  30.     private String password;
  31.  
  32.     @Value("${spring.redis.port}")
  33.     private int port;
  34.  
  35.     @Value("${spring.redis.timeout}")
  36.     private long timeout;
  37.  
  38.     @Value("${spring.redis.lettuce.shutdown-timeout}")
  39.     private long shutDownTimeout;
  40.  
  41.     @Value("${spring.redis.lettuce.pool.max-idle}")
  42.     private int maxIdle;
  43.  
  44.     @Value("${spring.redis.lettuce.pool.min-idle}")
  45.     private int minIdle;
  46.  
  47.     @Value("${spring.redis.lettuce.pool.max-active}")
  48.     private int maxActive;
  49.  
  50.     @Value("${spring.redis.lettuce.pool.max-wait}")
  51.     private long maxWait;
  52.  
  53.     Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
  54.  
  55.     @Bean
  56.     public LettuceConnectionFactory lettuceConnectionFactory() {
  57.         GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
  58.         genericObjectPoolConfig.setMaxIdle(maxIdle);
  59.         genericObjectPoolConfig.setMinIdle(minIdle);
  60.         genericObjectPoolConfig.setMaxTotal(maxActive);
  61.         genericObjectPoolConfig.setMaxWait(Duration.ofMillis(maxWait));
  62.         //每隔多少毫秒，空闲线程驱逐器关闭多余的空闲连接，且保持最少空闲连接可用，这个值最好设置大一点，否者影响性能
  63.         genericObjectPoolConfig.setTimeBetweenEvictionRuns(Duration.ofMillis(1000));
  64.         RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
  65.         redisStandaloneConfiguration.setDatabase(database);
  66.         redisStandaloneConfiguration.setHostName(host);
  67.         redisStandaloneConfiguration.setPort(port);
  68.         redisStandaloneConfiguration.setPassword(RedisPassword.of(password));
  69.         LettuceClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder()
  70.                 .commandTimeout(Duration.ofMillis(timeout))
  71.                 .shutdownTimeout(Duration.ofMillis(shutDownTimeout))
  72.                 .poolConfig(genericObjectPoolConfig)
  73.                 .build();
  74.  
  75.         LettuceConnectionFactory factory = new LettuceConnectionFactory(redisStandaloneConfiguration, clientConfig);
  76. //        factory.setShareNativeConnection(true);
  77. //        factory.setValidateConnection(false);
  78.         return factory;
  79.     }
  80.  
  81.     @Bean
  82.     public RedisSerializationContext redisSerializationContext() {
  83.         RedisSerializationContext.RedisSerializationContextBuilder builder = RedisSerializationContext.newSerializationContext();
  84.         builder.key(StringRedisSerializer.UTF_8);
  85.         builder.value(RedisSerializer.json());
  86.         builder.hashKey(StringRedisSerializer.UTF_8);
  87.         builder.hashValue(StringRedisSerializer.UTF_8);
  88.         return builder.build();
  89.     }
  90.  
  91.     @Bean
  92.     public ReactiveRedisTemplate<String, Object> reactiveRedisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
  93.         RedisSerializationContext serializationContext = redisSerializationContext();
  94.         return new ReactiveRedisTemplate(lettuceConnectionFactory,serializationContext);
  95.     }
  96.  
  97. }

四, 封装工具类

  1. package com.shuinfo.ods_poc.utils;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.data.redis.core.ReactiveRedisTemplate;
  6. import org.springframework.stereotype.Component;
  7. import reactor.core.publisher.Flux;
  8. import reactor.core.publisher.Mono;
  9.  
  10. import java.time.Duration;
  11.  
  12. @Component
  13. @Slf4j
  14. public class RedisUtils {
  15.     private static final Duration EXPIRED_DAY = Duration.ofDays(7);
  16.     private static final Duration EXPIRED_HOURS = Duration.ofHours(30);
  17.     private static final Duration EXPIRED_MINUTES = Duration.ofMinutes(1);
  18.     @Autowired
  19.     private ReactiveRedisTemplate reactiveRedisTemplate;
  20.  
  21.     public <T> void put(String key,T value){
  22.         reactiveRedisTemplate.opsForValue()
  23.             .set(key,value,EXPIRED_HOURS).subscribe((b->{}));
  24.     }
  25.  
  26.     public <T> Mono<T> get(String key){
  27.         return reactiveRedisTemplate.opsForValue()
  28.                 .get(key);
  29.     }
  30.  
  31.     public <T> void putList(String key, T value){
  32.         reactiveRedisTemplate.opsForSet().add(key,value).subscribe();
  33.     }
  34.  
  35.     public <T> Flux<T> getList(String key){
  36.         return reactiveRedisTemplate.opsForSet().members(key);
  37.     }
  38.  
  39.     public void remove(String key){
  40.         reactiveRedisTemplate.opsForSet().delete(key).subscribe();
  41.     }
  42.     public <T> void putZSet(String key, T value,String score){
  43.     reactiveRedisTemplate.opsForZSet().add(key,value,Double.valueOf(score))
  44.             .doOnError(e->{
  45.                 log.error(e.toString());
  46.             }).subscribe();
  47. }
  48.  
  49. public <T> Flux<T> getZSet(String key,Integer page,Integer size){
  50.     if(page == null || size == null){
  51.         return reactiveRedisTemplate.opsForZSet().rangeByScore(key,
  52.                 org.springframework.data.domain.Range.unbounded());
  53.     }
  54.      return reactiveRedisTemplate.opsForZSet().rangeByScore(key,
  55.              org.springframework.data.domain.Range.unbounded(),
  56.              org.springframework.data.redis.connection
  57.                      .RedisZSetCommands.Limit.limit()
  58.                      .offset((page - 1) * size).count(size)
  59.     );
  60. }
  61.  
  62. }

五, 使用

  1. //存缓存
  2. redisUtils.putList(RedisConstant.FIND_ORDER_BY_UID_AND_OTYPE
  3.         + req.getMemberCode()+req.getBusinessTypeCode(),req);
  4. redisUtils.putList(RedisConstant.FIND_ORDER_BY_SHOP_POS_INFO
  5.         + req.getShopCode()+req.getOrderTime()+req.getPosRefId()
  6.         + req.getReceiptNumber(),req);

六, 分布式锁(RedisTemplate)

  1. package com.shuinfo.ods_poc.utils;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.data.redis.connection.RedisStringCommands;
  5. import org.springframework.data.redis.connection.ReturnType;
  6. import org.springframework.data.redis.core.RedisCallback;
  7. import org.springframework.data.redis.core.RedisTemplate;
  8. import org.springframework.data.redis.core.types.Expiration;
  9. import org.springframework.stereotype.Component;
  10. import javax.annotation.Resource;
  11. import java.nio.charset.Charset;
  12. import java.util.concurrent.TimeUnit;
  13.  
  14. @Component
  15. @Slf4j
  16. public class RedisLockUtils {
  17.     @Resource
  18.     private RedisTemplate redisTemplate;
  19.  
  20.     private static final Long DEFAULT_EXPIRE_TIME = 1L;
  21.     private static final TimeUnit DEFAULT_UNIT = TimeUnit.SECONDS;
  22.     private static final String UNLOCK_LUA;
  23.     private static final String CHARCODE = "UTF-8";
  24.  
  25.     /**
  26.      * 释放锁脚本，原子操作
  27.      */
  28.     static {
  29.         StringBuilder sb = new StringBuilder();
  30.         sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
  31.         sb.append("then ");
  32.         sb.append("    return redis.call(\"del\",KEYS[1]) ");
  33.         sb.append("else ");
  34.         sb.append("    return 0 ");
  35.         sb.append("end ");
  36.         UNLOCK_LUA = sb.toString();
  37.     }
  38.  
  39.     /**
  40.      * 获取分布式锁，原子操作
  41.      * @param lockKey
  42.      * @param requestId 唯一ID, 可以使用UUID.randomUUID().toString();
  43.      * @param expire
  44.      * @param timeUnit
  45.      * @return
  46.      */
  47.     public boolean tryLock(String lockKey, String requestId, long expire, TimeUnit timeUnit) {
  48.         try{
  49.             RedisCallback<Boolean> callback = (connection) ->
  50.                  connection.set(lockKey.getBytes(Charset.forName(CHARCODE)),
  51.                         requestId.getBytes(Charset.forName(CHARCODE)),
  52.                         Expiration.seconds(timeUnit.toSeconds(expire)),
  53.                         RedisStringCommands.SetOption.SET_IF_ABSENT);
  54.             return (Boolean)redisTemplate.execute(callback);
  55.         } catch (Exception e) {
  56.             log.error("redis lock error.", e);
  57.         }
  58.         return false;
  59.     }
  60.     public boolean tryLock(String lockKey, String requestId) {
  61.         return tryLock( lockKey,  requestId,DEFAULT_EXPIRE_TIME,DEFAULT_UNIT);
  62.     }
  63.  
  64.     /**
  65.      * 释放锁
  66.      * @param lockKey
  67.      * @param requestId 唯一ID
  68.      * @return
  69.      */
  70.     public boolean releaseLock(String lockKey, String requestId) {
  71.         RedisCallback<Boolean> callback = (connection) ->
  72.                 connection.eval(UNLOCK_LUA.getBytes(), ReturnType.BOOLEAN ,
  73.                     1, lockKey.getBytes(Charset.forName(CHARCODE)),
  74.                     requestId.getBytes(Charset.forName(CHARCODE)));
  75.         return (Boolean)redisTemplate.execute(callback);
  76.     }
  77.  
  78.     /**
  79.      * 获取Redis锁的value值
  80.      * @param lockKey
  81.      * @return
  82.      */
  83.     public String get(String lockKey) {
  84.         try {
  85.             RedisCallback<String> callback = (connection) ->
  86.                  new String(connection.get(lockKey.getBytes()), Charset.forName(CHARCODE));
  87.             return (String)redisTemplate.execute(callback);
  88.         } catch (Exception e) {
  89.             log.error("get redis occurred an exception", e);
  90.         }
  91.         return null;
  92.     }
  93.  
  94. }

可参照阿里文档:

通过客户端程序连接Redis_云数据库 Redis 版(Redis)-阿里云帮助中心

注: redis的参数还需要关注redis服务端的参数,有些默认的参数可能实际并不适合

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/202744
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号