jwy
/
aeon_receiver
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

完善redis消费

jwy_csdj
review512jwy@163.com 2 days ago
parent
1f7338678d
commit
f1c6fe4aac
1 changed files with 150 additions and 82 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 232
      src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java

232
src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
View File

@ -11,12 +11,13 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Map;
@ -37,6 +38,19 @@ public class RedisStreamConsumer {
private static final Gson gson = new Gson();
/**
* 普通业务 Stream 最大保留消息数。
* Redis Stream 中，即使消息已经 ACK，也不会自动删除，如果不进行 Trim，Stream 会一直增长，最终占满 Redis 内存。
*/
private static final long STREAM_MAX_LEN = 100_000L; //100_000L;
/**
* 死信队列（DLQ）最大保留消息数。
* DLQ 中保存的是处理失败的消息，一般数量远少于正常 Stream，因此可以设置得更小。
*/
private static final long DLQ_MAX_LEN = 10_000L;
private static final int PARTITIONS = 16;
private static final String STREAM_PREFIX = "aeon_tcp_stream_";
private static final String GROUP = "aeon_tcp_stream_consumer_group";
@ -48,7 +62,7 @@ public class RedisStreamConsumer {
/**
* 读取 Pending 消息和 reclaim 时 批量处理消息数
*/
private static final int BATCH_SIZE = 50;
private static final int BATCH_SIZE = 200;
private volatile boolean running = true;
private String consumerName;
@ -56,7 +70,7 @@ public class RedisStreamConsumer {
/**
* 接收消息总数
*/
private final LongAdder receiveCounter = new LongAdder();
private final LongAdder processSubmitCounter = new LongAdder();
private final ExecutorService consumerExecutor =
Executors.newFixedThreadPool(PARTITIONS, new CustomizableThreadFactory("redis-stream-consumer-"));
@ -90,6 +104,18 @@ public class RedisStreamConsumer {
new CustomizableThreadFactory("redis-stats-")
);
/**
* 定时执行 Stream Trim 的线程池。
* 单线程即可，因为：
* 1、Trim 操作频率很低（例如每 30 分钟一次）
* 2、每次只是循环处理 16 个 Stream + 1 个 DLQ
* 3、不需要并发执行
*/
private final ScheduledExecutorService trimScheduler =
Executors.newSingleThreadScheduledExecutor(
new CustomizableThreadFactory("redis-trim-")
);
@PostConstruct
public void start() throws Exception {
consumerName = buildConsumerName();
@ -104,8 +130,6 @@ public class RedisStreamConsumer {
int partition = i;
consumerExecutor.submit(() -> {
// 启动时处理未 ack 消息
recoverPending(stream);
// 主消费循环
consumeLoop(stream, partition);
});
@ -121,9 +145,16 @@ public class RedisStreamConsumer {
// 定时统计接收次数
statsScheduler.scheduleAtFixedRate(() -> {
long count = receiveCounter.sum();
long count = processSubmitCounter.sum();
log.info("[F10] RedisStreamConsumer receive count={}", count);
}, 10, 10, TimeUnit.SECONDS);
trimScheduler.scheduleWithFixedDelay(
this::trimStreams,
10,
30,
TimeUnit.MINUTES
);
}
private String buildConsumerName() throws Exception {
@ -163,8 +194,20 @@ public class RedisStreamConsumer {
private void createGroup(String stream) {
try {
redisTemplate.opsForStream()
.createGroup(stream, ReadOffset.from("0-0"), GROUP);
// redisTemplate.opsForStream()
// .createGroup(stream, ReadOffset.from("0-0"), GROUP);
redisTemplate.execute((RedisCallback<Object>) connection -> {
connection.execute(
"XGROUP",
"CREATE".getBytes(StandardCharsets.UTF_8),
stream.getBytes(StandardCharsets.UTF_8),
GROUP.getBytes(StandardCharsets.UTF_8),
"0-0".getBytes(StandardCharsets.UTF_8),
"MKSTREAM".getBytes(StandardCharsets.UTF_8)
);
return null;
});
log.info("create group success stream={}", stream);
} catch (Exception e) {
Throwable cause = e.getCause();
@ -177,48 +220,6 @@ public class RedisStreamConsumer {
}
}
/**
* 启动时恢复未 ack 消息
*/
private void recoverPending(String stream) {
try {
// 分批次处理 Pending
while (true) {
PendingMessages pending = redisTemplate.opsForStream()
.pending(stream, GROUP, Range.unbounded(), BATCH_SIZE);
if (pending == null || pending.isEmpty()) break;
for (PendingMessage msg : pending) {
// claim 未 ack 消息
List<MapRecord<String, Object, Object>> claimed =
redisTemplate.opsForStream().claim(
stream,
GROUP,
consumerName,
IDLE_TIMEOUT,
msg.getId()
);
for (MapRecord<String, Object, Object> record : claimed) {
receiveCounter.increment();
// workerExecutor.submit(() -> process(record, stream, msg.getTotalDeliveryCount()));
submitTask(() ->
process(record, stream, msg.getTotalDeliveryCount())
);
}
}
// 如果少于 BATCH_SIZE，说明已处理完
if (pending.size() < BATCH_SIZE) break;
}
log.info("recoverPending stream={}", stream);
} catch (Exception e) {
log.error("recoverPending error stream={}", stream, e);
}
}
private void consumeLoop(String stream, int partition) {
while (running) {
try {
@ -233,7 +234,7 @@ public class RedisStreamConsumer {
if (messages == null || messages.isEmpty()) continue;
receiveCounter.add(messages.size());
processSubmitCounter.add(messages.size());
for (MapRecord<String, Object, Object> message : messages) {
// workerExecutor.submit(() -> process(message, stream, 1));
@ -323,49 +324,108 @@ public class RedisStreamConsumer {
*/
private void reclaimIdleMessages() {
if (workerExecutor.getQueue().size() > WORK_QUEUE_LIMIT) {
log.warn("skip reclaim, worker queue busy");
log.warn("skip reclaim, worker queue busy, queueSize={}", workerExecutor.getQueue().size());
return;
}
for (int i = 1; i <= PARTITIONS; i++) {
String stream = STREAM_PREFIX + i;
try {
while (true) {
PendingMessages pending =
redisTemplate.opsForStream().pending(stream, GROUP, Range.unbounded(), BATCH_SIZE);
PendingMessages pending =
redisTemplate.opsForStream().pending(stream, GROUP, Range.unbounded(), BATCH_SIZE);
if (pending == null || pending.isEmpty()) {
continue;
}
for (PendingMessage p : pending) {
if (p.getElapsedTimeSinceLastDelivery().compareTo(IDLE_TIMEOUT) < 0) {
continue;
}
List<MapRecord<String,Object,Object>> claimed =
redisTemplate.opsForStream()
.claim(
stream,
GROUP,
consumerName,
IDLE_TIMEOUT,
p.getId()
);
if (claimed == null || claimed.isEmpty()) {
continue;
}
for(MapRecord<String, Object, Object> r : claimed){
processSubmitCounter.increment();
// workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount()));
submitTask(() ->
process(r, stream, p.getTotalDeliveryCount())
);
}
}
} catch(Exception e){
log.error("reclaim error stream={}",stream,e
);
}
}
}
if (pending == null || pending.isEmpty()) break;
/**
* 定时裁剪所有 Stream。
* 当前系统包括：
*
* 业务 Stream：
* aeon_tcp_stream_1
* aeon_tcp_stream_2
* ...
* aeon_tcp_stream_16
*
* 死信队列：
* aeon_tcp_stream_dlq
*
* 主 Stream 和 DLQ 分开裁剪，
* 可以使用不同的最大保留数量。
*/
private void trimStreams() {
for (PendingMessage p : pending) {
if (p.getElapsedTimeSinceLastDelivery().compareTo(IDLE_TIMEOUT) < 0) continue;
// 裁剪所有业务 Stream
for (int i = 1; i <= PARTITIONS; i++) {
trimStream(STREAM_PREFIX + i, STREAM_MAX_LEN);
}
List<MapRecord<String, Object, Object>> claimed =
redisTemplate.opsForStream().claim(
stream,
GROUP,
consumerName,
IDLE_TIMEOUT,
p.getId()
);
// 裁剪死信队列
trimStream(DLQ_STREAM, DLQ_MAX_LEN);
}
for (MapRecord<String, Object, Object> r : claimed) {
receiveCounter.increment();
// workerExecutor.submit(() -> process(r, stream, p.getTotalDeliveryCount()));
submitTask(() ->
process(r, stream, p.getTotalDeliveryCount())
);
}
}
/**
* 裁剪指定 Stream。
* Redis 命令：XTRIM stream MAXLEN ~ maxLen
* 例如：XTRIM aeon_tcp_stream_1 MAXLEN ~ 100000 表示：当 Stream 超过约10万条时，Redis 自动删除最旧的数据。
*
* "~" 表示 Approximate Trim（近似裁剪）。
*
* 为什么不用精确裁剪？
* 精确裁剪：MAXLEN 1000000，Redis 每次都会精确删除到1000000条，CPU 消耗较高。
* 近似裁剪：MAXLEN ~ 1000000，Redis 会在内部合适的时机进行裁剪，性能远高于精确裁剪。
*
* @param streamName Stream 名称
* @param maxLen 最大保留消息数量
*/
private void trimStream(String streamName, long maxLen) {
try {
redisTemplate.execute((RedisCallback<Void>) connection -> {
connection.streamCommands().xTrim(
streamName.getBytes(StandardCharsets.UTF_8),
maxLen,
true
);
return null;
});
if (pending.size() < BATCH_SIZE) break;
}
} catch (Exception e) {
log.error("[F10] reclaim error stream={}", stream, e);
}
log.info("Trim stream success, stream={}, maxLen={}", streamName, maxLen);
} catch (Exception e) {
log.error("Trim stream failed, stream={}", streamName, e);
}
}
public long getReceiveCount() {
return receiveCounter.sum();
return processSubmitCounter.sum();
}
private void sleep(long ms) {
@ -385,11 +445,13 @@ public class RedisStreamConsumer {
workerExecutor.shutdown();
reclaimScheduler.shutdown();
statsScheduler.shutdown();
trimScheduler.shutdown();
boolean consumerTerminated = consumerExecutor.awaitTermination(10, TimeUnit.SECONDS);
boolean workerTerminated = workerExecutor.awaitTermination(10, TimeUnit.SECONDS);
boolean reclaimTerminated = reclaimScheduler.awaitTermination(10, TimeUnit.SECONDS);
boolean statsTerminated = statsScheduler.awaitTermination(10, TimeUnit.SECONDS);
boolean trimTerminated = trimScheduler.awaitTermination(10, TimeUnit.SECONDS );
if (!consumerTerminated) {
consumerExecutor.shutdownNow();
@ -403,11 +465,17 @@ public class RedisStreamConsumer {
if (!statsTerminated) {
statsScheduler.shutdownNow();
}
if (!trimTerminated) {
trimScheduler.shutdownNow();
}
log.info("RedisStreamConsumer shutdown done, consumerTerminated={}, workerTerminated={}, reclaimTerminated={}, statsTerminated={}",
log.info(
"RedisStreamConsumer shutdown done, consumerTerminated={}, workerTerminated={}, reclaimTerminated={}, statsTerminated={}, trimTerminated={}",
consumerTerminated,
workerTerminated,
reclaimTerminated,
statsTerminated);
statsTerminated,
trimTerminated
);
}
}
Write Preview
Loading…
Cancel
Save