From 1849b0f37ee05e41e774d6af53ae642eb4bbee6a Mon Sep 17 00:00:00 2001 From: thetumbled Date: Thu, 19 Jun 2025 12:02:20 +0800 Subject: [PATCH 1/3] introduce delayedMessagesCount to reduce cpu usage. --- .../delayed/InMemoryDelayedDeliveryTracker.java | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java index a48ed4161386d..802f7146ebede 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java @@ -30,6 +30,8 @@ import java.util.NavigableSet; import java.util.TreeSet; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicLong; + import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.Position; @@ -64,6 +66,9 @@ public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTrack // The bit count to trim to reduce memory occupation. private final int timestampPrecisionBitCnt; + // 添加一个原子类型的计数器 + private final AtomicLong delayedMessagesCount = new AtomicLong(0); + InMemoryDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer, long tickTimeMillis, boolean isDelayedDeliveryDeliverAtTimeStrict, @@ -125,6 +130,8 @@ public boolean addMessage(long ledgerId, long entryId, long deliverAt) { delayedMessageMap.computeIfAbsent(timestamp, k -> new Long2ObjectRBTreeMap<>()) .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap()) .add(entryId); + delayedMessagesCount.incrementAndGet(); + updateTimer(); checkAndUpdateHighest(deliverAt); @@ -183,6 +190,7 @@ public NavigableSet getScheduledMessages(int maxMessages) { positions.add(PositionFactory.create(ledgerId, entryId)); }); n -= cardinality; + delayedMessagesCount.addAndGet(-cardinality); ledgerIdToDelete.add(ledgerId); } else { long[] entryIdsArray = entryIds.toArray(); @@ -190,6 +198,7 @@ public NavigableSet getScheduledMessages(int maxMessages) { positions.add(PositionFactory.create(ledgerId, entryIdsArray[i])); entryIds.removeLong(entryIdsArray[i]); } + delayedMessagesCount.addAndGet(-n); n = 0; } if (n <= 0) { @@ -221,14 +230,13 @@ public NavigableSet getScheduledMessages(int maxMessages) { @Override public CompletableFuture clear() { this.delayedMessageMap.clear(); + this.delayedMessagesCount.set(0); return CompletableFuture.completedFuture(null); } @Override public long getNumberOfDelayedMessages() { - return delayedMessageMap.values().stream().mapToLong( - ledgerMap -> ledgerMap.values().stream().mapToLong( - Roaring64Bitmap::getLongCardinality).sum()).sum(); + return delayedMessagesCount.get(); } /** From ee0cf94b5416511bb92dba3783a259b0ea80b930 Mon Sep 17 00:00:00 2001 From: thetumbled Date: Thu, 19 Jun 2025 12:07:42 +0800 Subject: [PATCH 2/3] update. --- .../pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java index 802f7146ebede..5dfe451664236 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java @@ -66,7 +66,7 @@ public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTrack // The bit count to trim to reduce memory occupation. private final int timestampPrecisionBitCnt; - // 添加一个原子类型的计数器 + // Count of delayed messages in the tracker. private final AtomicLong delayedMessagesCount = new AtomicLong(0); InMemoryDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer, From 177b3be5faa78a7fd1de8e3aeba6632258015992 Mon Sep 17 00:00:00 2001 From: thetumbled Date: Thu, 19 Jun 2025 15:30:11 +0800 Subject: [PATCH 3/3] fix check style. --- .../pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java index 5dfe451664236..900896cdf47d2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java @@ -31,7 +31,6 @@ import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicLong; - import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.Position;