解决2个问题:

1、捕获异常调整到循环内,防止因为redis代码抛出异常程序中断。
2、新增销毁方法,在程序优雅结束进程后,退出while循环。
master
xzxiaoshan 2 months ago
parent ca4d91a8c2
commit bee9e24085

@ -14,6 +14,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
@ -22,7 +23,6 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Consumer; import java.util.function.Consumer;
/** /**
*
* Redis * Redis
* *
* @author xiaoxiamao * @author xiaoxiamao
@ -43,13 +43,18 @@ public class RedisReceiver implements MessageReceiver {
@Autowired @Autowired
private ConsumeService consumeService; private ConsumeService consumeService;
/**
* 线
*/
private ScheduledExecutorService scheduler;
/** /**
* 线 * 线
*/ */
@PostConstruct @PostConstruct
public void init() { public void init() {
// 创建调度线程池 // 创建调度线程池
ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(2, this.scheduler = new ScheduledThreadPoolExecutor(2,
r -> new Thread(r, "RedisReceiverThread")); r -> new Thread(r, "RedisReceiverThread"));
// 定时调度 // 定时调度
scheduler.scheduleWithFixedDelay(this::receiveSendMessage, 0, 1, TimeUnit.SECONDS); scheduler.scheduleWithFixedDelay(this::receiveSendMessage, 0, 1, TimeUnit.SECONDS);
@ -80,23 +85,39 @@ public class RedisReceiver implements MessageReceiver {
/** /**
* *
* * <p>
* *
* *
* @param topic * @param topic
* @param consumer * @param consumer
*/ */
private void receiveMessage(String topic, Consumer<String> consumer) { private void receiveMessage(String topic, Consumer<String> consumer) {
try { while (true) {
while (true) { try {
// 阻塞操作减少CPUIO消耗 // 阻塞操作减少CPUIO消耗
Optional<String> message = Optional.ofNullable( Optional<String> message = Optional.ofNullable(
stringRedisTemplate.opsForList().rightPop(topic, 20, TimeUnit.SECONDS)); stringRedisTemplate.opsForList().rightPop(topic, 20, TimeUnit.SECONDS));
message.ifPresent(consumer); message.ifPresent(consumer);
} catch (Exception e) {
log.error("RedisReceiver#receiveMessage Error receiving messages from Redis topic {}: {}",
topic, e.getMessage());
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException ex) {
log.error("RedisReceiver#receiveMessage interrupted: {}", e.getMessage());
Thread.currentThread().interrupt();
break;
}
} }
} catch (Exception e) {
log.error("RedisReceiver#receiveMessage Error receiving messages from Redis topic {}: {}",
topic, e.getMessage());
} }
} }
/**
*
*/
@PreDestroy
public void onDestroy() {
scheduler.shutdown();
}
} }

Loading…
Cancel
Save