对于MQ消费使用redis缓冲限流处理

一、好言

事实是这样，假如你不懂我，那错的永远是我，不必惊讶，连解释都是多余。

二、背景

由于消费MQ，会存在有就消费，可能会有并发的存在，在后台引起的问题，所以需要多MQ的消费做处理，然后我们将MQ消息消费后，丢进redsi,然后从开启线程，从redis中取数据进行消费，下面给出的是set和队列的两种方式。

三、内容

3.1 SET的处理方式

 @PostConstruct
public void dealTask() throws Exception {
        ScheduledExecutorService es = Executors.newScheduledThreadPool(100);
        es.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                logger.info("================");
                Long flag = stringRedisTemplate.opsForValue().increment(RedisConstant.TASK_SWITCH_POWER,1);
                if(flag == 1 ){//这个标识用来控制集群，每次只有等于1的进入，取完数据或者没有数据都需要将该
值重新置为0
                    ZSetOperations<String, VO> zSetOperations = objectredisTemplate.opsForZSet();
                    Long count = zSetOperations.size(RedisConstant.NAME_TASK_KEY);
                    if(count > 0){
                        Set<VO> zsetRangeData = zSetOperations.range(RedisConstant.NAME_TASK_KEY,0L, 
count - 1);
                        zSetOperations.remove(RedisConstant.NAME_TASK_KEY, 
zsetRangeData.toArray());
                        stringRedisTemplate.opsForValue().set(RedisConstant.TASK_SWITCH_POWER, "0");
                        Iterator iterable = zsetRangeData.iterator();
                        while (iterable.hasNext()){
                            try {
                               //TODO处理业务
                            }catch (Exception e){
                                //出异常在把该信息添加进去set
                                zSetOperations.add(RedisConstant.NAME_TASK_KEY, 
ZSetUtil.converToSet(vo));
                            }
                        }
                    }else{
                        stringRedisTemplate.opsForValue().set(RedisConstant.TASK_SWITCH_POWER, "0");
                    }
                }
            }
        },0,1, TimeUnit.SECONDS);
    }

接收端接收消息然后丢入redis，当然你可以是指redis大小数量，超过上限则可以直接丢弃

               ZSetOperations<String, VO> zSet = redisTemplate.opsForZSet();
                if(zSet.size(RedisConstant.NAME_TASK_KEY) + 1  >= 
MAX_CHANNEL_MQ_TASK_SIZE){
                    logger.info("到达上限,丢弃消息 = "+messgage);
                    return;
                }
                zSet.add(RedisConstant.NAME_TASK_KEY, ZSetUtil.converToSet(messgage));

ZSetUtil.java

   public static Set<ZSetOperations.TypedTuple<VO>> converToSet(VO data){
        Set<ZSetOperations.TypedTuple<VO>> set = new HashSet<>(1);
        ZSetOperations.TypedTuple typedTuple;
        Long score = System.currentTimeMillis();
        typedTuple = new DefaultTypedTuple(data, score.doubleValue());
        set.add(typedTuple);
        return set;
    }

3.2、队列方式

队列方式先push然后pop出来，每次处理一条数据。

消息接收端

 ListOperations listOperations = redisTemplate.opsForList();
 redisTemplate.opsForList().leftPush(RedisConstant.NAME_TASK_KEY,msg);

开启定时任务从redis中取数据

    @PostConstruct
    public void dealNameTask(){
     ScheduledExecutorService es = Executors.newScheduledThreadPool(10);
        es.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                ListOperations listOperations = objectredisTemplate.opsForList();
                VO vo = (VO)listOperations.rightPop(RedisConstant.NAME_TASK_KEY);
                try {
                    if(null != vo){
                        //Todo处理业务
                    }else{
                        Thread.sleep(1000);
                    }
                }catch (Exception e){
                    //出现异常在push进redis
                    listOperations.leftPush(RedisConstant.NAME_TASK_ZSET_KEY,vo);
                }
            }
        },0,1, TimeUnit.SECONDS);
    }

相对来说，push会简单多了，只是set可以排序，并且也可以按照顺序来取。