[improve][broker] Avoid share same random in multi-threads due to per…

…formance issue (apache#18660)
lifepuzzlefun · Jan 10, 2023 · bbf8e26 · bbf8e26
1 parent 4f7cacc
commit bbf8e26
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 9 deletions.
diff --git a/...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java b/...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
@@ -20,8 +20,8 @@
 
 import com.carrotsearch.hppc.ObjectHashSet;
 import com.carrotsearch.hppc.ObjectSet;
-import java.util.Random;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
@@ -45,8 +45,6 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
                     .newUpdater(AbstractDispatcherMultipleConsumers.class, "isClosed");
     private volatile int isClosed = FALSE;
 
-    private Random random = new Random(42);
-
     protected AbstractDispatcherMultipleConsumers(Subscription subscription, ServiceConfiguration serviceConfig) {
         super(subscription, serviceConfig);
     }
@@ -157,7 +155,7 @@ public Consumer getRandomConsumer() {
             return null;
         }
 
-        return consumerList.get(random.nextInt(consumerList.size()));
+        return consumerList.get(ThreadLocalRandom.current().nextInt(consumerList.size()));
     }
 
 

diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -49,6 +49,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -392,7 +393,6 @@ public static void main(String[] args) throws Exception {
                     msgRatePerThread,
                     payloadByteList,
                     payloadBytes,
-                    random,
                     doneLatch
                 );
             });
@@ -529,7 +529,6 @@ private static void runProducer(int producerId,
                                     int msgRate,
                                     List<byte[]> payloadByteList,
                                     byte[] payloadBytes,
-                                    Random random,
                                     CountDownLatch doneLatch) {
         PulsarClient client = null;
         try {
@@ -626,9 +625,10 @@ private static void runProducer(int producerId,
                     if (arguments.payloadFilename != null) {
                         if (messageFormatter != null) {
                             payloadData = messageFormatter.formatMessage(arguments.producerName, totalSent,
-                                    payloadByteList.get(random.nextInt(payloadByteList.size())));
+                                    payloadByteList.get(ThreadLocalRandom.current().nextInt(payloadByteList.size())));
                         } else {
-                            payloadData = payloadByteList.get(random.nextInt(payloadByteList.size()));
+                            payloadData = payloadByteList.get(
+                                    ThreadLocalRandom.current().nextInt(payloadByteList.size()));
                         }
                     } else {
                         payloadData = payloadBytes;
@@ -656,7 +656,7 @@ private static void runProducer(int producerId,
                     }
                     //generate msg key
                     if (msgKeyMode == MessageKeyGenerationMode.random) {
-                        messageBuilder.key(String.valueOf(random.nextInt()));
+                        messageBuilder.key(String.valueOf(ThreadLocalRandom.current().nextInt()));
                     } else if (msgKeyMode == MessageKeyGenerationMode.autoIncrement) {
                         messageBuilder.key(String.valueOf(totalSent));
                     }