一、好言
事实是这样，假如你不懂我，那错的永远是我，不必惊讶，连解释都是多余。
二、背景
由于消费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可以排序，并且也可以按照顺序来取。
