文档章节

SpringBoot2.0高级案例(07) :整合Redis集群 ,实现消息队列场景

知了一笑
 知了一笑
发布于 07/21 23:03
字数 934
阅读 113
收藏 1
本文源码
GitHub地址:知了一笑
https://github.com/cicadasmile/middle-ware-parent

一、Redis集群简介

1、RedisCluster概念

Redis的分布式解决方案,在3.0版本后推出的方案,有效地解决了Redis分布式的需求,当一个服务宕机可以快速的切换到另外一个服务。redis cluster主要是针对海量数据+高并发+高可用的场景。

二、与SpringBoot2.0整合

1、核心依赖

<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>

2、核心配置

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

3、参数渲染类

@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方法
}

4、集群配置文件

@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();
    }
}

5、配置Redis模板类

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

三、模拟队列场景案例

生产者消费者模式:客户端监听消息队列,消息达到,消费者马上消费,如果消息队列里面没有消息,那么消费者就继续监听。基于Redis的LPUSH(BLPUSH)把消息入队,用 RPOP(BRPOP)获取消息的模式。

1、加锁解锁工具

@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();
        }
    }
}

2、消息消费

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";
    }
}

3、消息监听器

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();
            }
        }
    }
}

4、消息生产者

@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();
        }
    }
}

5、场景测试接口

@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" ;
    }
}

四、源代码地址

GitHub地址:知了一笑
https://github.com/cicadasmile/middle-ware-parent
码云地址:知了一笑
https://gitee.com/cicadasmile/middle-ware-parent

© 著作权归作者所有

知了一笑

知了一笑

粉丝 111
博文 45
码字总数 44492
作品 0
杭州
私信 提问
小柒2012/spring-boot-seckill

分布式秒杀系统 开发环境 JDK1.7、Maven、Mysql、Eclipse、SpringBoot1.5.10、zookeeper3.4.6、kafka_2.11、redis-2.8.4、curator-2.10.0 友情提示 由于工作原因,项目正在完善中(仅供参考)...

小柒2012
2018/05/19
0
0
SpringBoot开发案例从0到1构建分布式秒杀系统

前言 最近,被推送了不少秒杀架构的文章,忙里偷闲自己也总结了一下互联网平台秒杀架构设计,当然也借鉴了不少同学的思路。俗话说,脱离案例讲架构都是耍流氓,最终使用SpringBoot模拟实现了...

小柒2012
2018/05/16
0
0
Redis高级特性介绍及实例分析

本文将为大家介绍Redis的一些高级特性以及结合一个具体的实际案例来对Redis进行设计分析。 Redis基础类型回顾 String Redis中最基本,也是最简单的数据类型。注意,VALUE既可以是简单的Strin...

zfz_linux_boy
2017/02/26
0
0
架构师必备词汇和知识点

01 高可用 负载均衡(负载均衡算法) 反向代理 服务隔离 服务限流 服务降级(自动优雅降级) 失效转移 超时重试(代理超时、容器超时、前端超时、中间件超时、数据库超时、NoSql超时) 回滚机...

t4i2b10X4c22nF6A
2017/11/24
0
0
简明实用:Redis 高级特性与案例介绍

本文将为大家介绍Redis的一些高级特性以及结合一个具体的实际案例来对Redis进行设计分析。 Redis基础类型回顾 String Redis中最基本,也是最简单的数据类型。注意,VALUE既可以是简单的Strin...

高效运维
2018/07/18
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Spring boot 静态资源访问

0. 两个配置 spring.mvc.static-path-patternspring.resources.static-locations 1. application中需要先行的两个配置项 1.1 spring.mvc.static-path-pattern 这个配置项是告诉springboo......

moon888
今天
3
0
hash slot(虚拟桶)

在分布式集群中,如何保证相同请求落到相同的机器上,并且后面的集群机器可以尽可能的均分请求,并且当扩容或down机的情况下能对原有集群影响最小。 round robin算法:是把数据mod后直接映射...

李朝强
今天
4
0
Kafka 原理和实战

本文首发于 vivo互联网技术 微信公众号 https://mp.weixin.qq.com/s/bV8AhqAjQp4a_iXRfobkCQ 作者简介:郑志彬,毕业于华南理工大学计算机科学与技术(双语班)。先后从事过电子商务、开放平...

vivo互联网技术
今天
19
0
java数据类型

基本类型: 整型:Byte,short,int,long 浮点型:float,double 字符型:char 布尔型:boolean 引用类型: 类类型: 接口类型: 数组类型: Byte 1字节 八位 -128 -------- 127 short 2字节...

audience_1
今天
9
0
太全了|万字详解Docker架构原理、功能及使用

一、简介 1、了解Docker的前生LXC LXC为Linux Container的简写。可以提供轻量级的虚拟化,以便隔离进程和资源,而且不需要提供指令解释机制以及全虚拟化的其他复杂性。相当于C++中的NameSpa...

Java技术剑
今天
27
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部