SpringBoot2.0 整合 Redis集羣 ,實現消息隊列場景

本文源碼
GitHub地址：知了一笑
https://github.com/cicadasmile/middle-ware-parent

1、Redis集羣簡介

一、RedisCluster概念

Redis的分佈式解決方案，在3.0版本後推出的方案，有效地解決了Redis分佈式的需求，當一個服務宕機能夠快速的切換到另一個服務。redis cluster主要是針對海量數據+高併發+高可用的場景。

2、與SpringBoot2.0整合

一、核心依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>${spring-boot.version}</version>
</dependency>
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>${redis-client.version}</version>
</dependency>

二、核心配置

spring:
  # Redis 集羣
  redis:
    sentinel:
      # sentinel 配置
      master: mymaster
      nodes: 192.168.0.127:26379
      maxTotal: 60
      minIdle: 10
      maxWaitMillis: 10000
      testWhileIdle: true
      testOnBorrow: true
      testOnReturn: false
      timeBetweenEvictionRunsMillis: 10000

三、參數渲染類

@ConfigurationProperties(prefix = "spring.redis.sentinel")
public class RedisParam {
    private String nodes ;
    private String master ;
    private Integer maxTotal ;
    private Integer minIdle ;
    private Integer maxWaitMillis ;
    private Integer timeBetweenEvictionRunsMillis ;
    private boolean testWhileIdle ;
    private boolean testOnBorrow ;
    private boolean testOnReturn ;
    // 省略GET和SET方法
}

四、集羣配置文件

@Configuration
@EnableConfigurationProperties(RedisParam.class)
public class RedisPool {
    @Resource
    private RedisParam redisParam ;
    @Bean("jedisSentinelPool")
    public JedisSentinelPool getRedisPool (){
        Set<String> sentinels = new HashSet<>();
        sentinels.addAll(Arrays.asList(redisParam.getNodes().split(",")));
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(redisParam.getMaxTotal());
        poolConfig.setMinIdle(redisParam.getMinIdle());
        poolConfig.setMaxWaitMillis(redisParam.getMaxWaitMillis());
        poolConfig.setTestWhileIdle(redisParam.isTestWhileIdle());
        poolConfig.setTestOnBorrow(redisParam.isTestOnBorrow());
        poolConfig.setTestOnReturn(redisParam.isTestOnReturn());
        poolConfig.setTimeBetweenEvictionRunsMillis(redisParam.getTimeBetweenEvictionRunsMillis());
        JedisSentinelPool redisPool = new JedisSentinelPool(redisParam.getMaster(), sentinels, poolConfig);
        return redisPool;
    }
    @Bean
    SpringUtil springUtil() {
        return new SpringUtil();
    }
    @Bean
    RedisListener redisListener() {
        return new RedisListener();
    }
}

五、配置Redis模板類

@Configuration
public class RedisConfig {
    @Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
        StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
        stringRedisTemplate.setConnectionFactory(factory);
        return stringRedisTemplate;
    }
}

3、模擬隊列場景案例

生產者消費者模式：客戶端監聽消息隊列，消息達到，消費者立刻消費，若是消息隊列裏面沒有消息，那麼消費者就繼續監聽。基於Redis的LPUSH（BLPUSH）把消息入隊，用 RPOP（BRPOP）獲取消息的模式。

一、加鎖解鎖工具

@Component
public class RedisLock {
    private static String keyPrefix = "RedisLock:";
    @Resource
    private JedisSentinelPool jedisSentinelPool;
    public boolean addLock(String key, long expire) {
        Jedis jedis = null;
        try {
            jedis = jedisSentinelPool.getResource();
            /*
             * nxxx的值只能取NX或者XX，若是取NX，則只有當key不存在是才進行set，若是取XX，則只有當key已經存在時才進行set
             * expx的值只能取EX或者PX，表明數據過時時間的單位，EX表明秒，PX表明毫秒。
             */
            String value = jedis.set(keyPrefix + key, "1", "nx", "ex", expire);
            return value != null;
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            if (jedis != null) jedis.close();
        }
        return false;
    }
    public void removeLock(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisSentinelPool.getResource();
            jedis.del(keyPrefix + key);
        } finally {
            if (jedis != null) jedis.close();
        }
    }
}

二、消息消費

1）封裝接口

public interface RedisHandler  {
    /**
     * 隊列名稱
     */
    String queueName();

    /**
     * 隊列消息內容
     */
    String consume (String msgBody);
}

2）接口實現

@Component
public class LogAListen implements RedisHandler {
    private static final Logger LOG = LoggerFactory.getLogger(LogAListen.class) ;
    @Resource
    private RedisLock redisLock;
    @Override
    public String queueName() {
        return "LogA-key";
    }
    @Override
    public String consume(String msgBody) {
        // 加鎖，防止消息重複投遞
        String lockKey = "lock-order-uuid-A";
        boolean lock = false;
        try {
            lock = redisLock.addLock(lockKey, 60);
            if (!lock) {
                return "success";
            }
            LOG.info("LogA-key == >>" + msgBody);
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            if (lock) {
                redisLock.removeLock(lockKey);
            }
        }
        return "success";
    }
}

三、消息監聽器

public class RedisListener implements InitializingBean {
    /**
     * Redis 集羣
     */
    @Resource
    private JedisSentinelPool jedisSentinelPool;
    private List<RedisHandler> handlers = null;
    private ExecutorService product = null;
    private ExecutorService consumer = null;
    /**
     * 初始化配置
     */
    @Override
    public void afterPropertiesSet() {
        handlers = SpringUtil.getBeans(RedisHandler.class) ;
        product = new ThreadPoolExecutor(10,15,60 * 3,
                TimeUnit.SECONDS,new SynchronousQueue<>());
        consumer = new ThreadPoolExecutor(10,15,60 * 3,
                TimeUnit.SECONDS,new SynchronousQueue<>());
        for (RedisHandler redisHandler : handlers){
            product.execute(() -> {
                redisTask(redisHandler);
            });
        }
    }
    /**
     * 隊列監聽
     */
    public void redisTask (RedisHandler redisHandler){
        Jedis jedis = null ;
        while (true){
            try {
                jedis = jedisSentinelPool.getResource() ;
                List<String> msgBodyList = jedis.brpop(0, redisHandler.queueName());
                if (msgBodyList != null && msgBodyList.size()>0){
                    consumer.execute(() -> {
                        redisHandler.consume(msgBodyList.get(1)) ;
                    });
                }
            } catch (Exception e){
                e.printStackTrace();
            } finally {
                if (jedis != null) jedis.close();
            }
        }
    }
}

四、消息生產者

@Service
public class RedisServiceImpl implements RedisService {
    @Resource
    private JedisSentinelPool jedisSentinelPool;
    @Override
    public void saveQueue(String queueKey, String msgBody) {
        Jedis jedis = null;
        try {
            jedis = jedisSentinelPool.getResource();
            jedis.lpush(queueKey,msgBody) ;
        } catch (Exception e){
          e.printStackTrace();
        } finally {
            if (jedis != null) jedis.close();
        }
    }
}

五、場景測試接口

@RestController
public class RedisController {
    @Resource
    private RedisService redisService ;
    /**
     * 隊列推消息
     */
    @RequestMapping("/saveQueue")
    public String saveQueue (){
        MsgBody msgBody = new MsgBody() ;
        msgBody.setName("LogAModel");
        msgBody.setDesc("描述");
        msgBody.setCreateTime(new Date());
        redisService.saveQueue("LogA-key", JSONObject.toJSONString(msgBody));
        return "success" ;
    }
}

4、源代碼地址

GitHub地址：知了一笑
https://github.com/cicadasmile
碼雲地址：知了一笑
https://gitee.com/cicadasmile