Skip to content

pulsar
pulsar

apache / pulsar

通知 129
Star 40
Fork 3
pulsar
pulsar

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 b7522dfb 编写于 作者: S Sijie Guo 提交者： xiaolong.ran
浏览文件
操作

[pulsar-perf] Introduce number of threads in perf producer program (#5036) 

*Motivation*

Allow spinning up multiple threads for running the perf-producer.

*Modifications*

- Introduce a new option for configuring multiple threads
- Each thread has one separate pulsar client

(cherry picked from commit 0b1e3b37)
上级 6b3c7b73
隐藏空白更改
内联 并排
Showing with 201 addition and 157 deletion
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
浏览文件 @ b7522dfb
......@@ -45,11 +45,11 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
import org.HdrHistogram.Histogram;
......@@ -63,11 +63,15 @@ import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* A client program to test pulsar producer performance.
*/
public class PerformanceProducer {
private static final ExecutorService executor = Executors
......@@ -93,6 +97,9 @@ public class PerformanceProducer {
@Parameter(description = "persistent://prop/ns/my-topic", required = true)
public List<String> topics;
@Parameter(names = { "-threads", "--num-test-threads" }, description = "Number of test threads")
public int numTestThreads = 1;
@Parameter(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
public int msgRate = 100;
......@@ -173,6 +180,32 @@ public class PerformanceProducer {
public long delay = 0;
}
static class EncKeyReader implements CryptoKeyReader {
private static final long serialVersionUID = 7235317430835444498L;
final String encKeyName;
final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
EncKeyReader(String encKeyName, byte[] value) {
this.encKeyName = encKeyName;
keyInfo.setKey(value);
}
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
if (keyName.equals(encKeyName)) {
return keyInfo;
}
return null;
}
@Override
public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMeta) {
return null;
}
}
public static void main(String[] args) throws Exception {
final Arguments arguments = new Arguments();
......@@ -255,85 +288,6 @@ public class PerformanceProducer {
}
}
// Now processing command line arguments
String prefixTopicName = arguments.topics.get(0);
List<Future<Producer<byte[]>>> futures = Lists.newArrayList();
ClientBuilder clientBuilder = PulsarClient.builder() //
.serviceUrl(arguments.serviceURL) //
.connectionsPerBroker(arguments.maxConnections) //
.ioThreads(Runtime.getRuntime().availableProcessors()) //
.statsInterval(arguments.statsIntervalSeconds, TimeUnit.SECONDS) //
.tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);
if (isNotBlank(arguments.authPluginClassName)) {
clientBuilder.authentication(arguments.authPluginClassName, arguments.authParams);
}
class EncKeyReader implements CryptoKeyReader {
EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
EncKeyReader(byte[] value) {
keyInfo.setKey(value);
}
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
if (keyName.equals(arguments.encKeyName)) {
return keyInfo;
}
return null;
}
@Override
public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMeta) {
return null;
}
}
PulsarClient client = clientBuilder.build();
ProducerBuilder<byte[]> producerBuilder = client.newProducer() //
.sendTimeout(0, TimeUnit.SECONDS) //
.compressionType(arguments.compression) //
.maxPendingMessages(arguments.maxOutstanding) //
.maxPendingMessagesAcrossPartitions(arguments.maxPendingMessagesAcrossPartitions)
// enable round robin message routing if it is a partitioned topic
.messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
if (arguments.batchTimeMillis == 0.0) {
producerBuilder.enableBatching(false);
} else {
long batchTimeUsec = (long) (arguments.batchTimeMillis * 1000);
producerBuilder.batchingMaxPublishDelay(batchTimeUsec, TimeUnit.MICROSECONDS)
.enableBatching(true);
}
// Block if queue is full else we will start seeing errors in sendAsync
producerBuilder.blockIfQueueFull(true);
if (arguments.encKeyName != null) {
producerBuilder.addEncryptionKey(arguments.encKeyName);
byte[] pKey = Files.readAllBytes(Paths.get(arguments.encKeyFile));
EncKeyReader keyReader = new EncKeyReader(pKey);
producerBuilder.cryptoKeyReader(keyReader);
}
for (int i = 0; i < arguments.numTopics; i++) {
String topic = (arguments.numTopics == 1) ? prefixTopicName : String.format("%s-%d", prefixTopicName, i);
log.info("Adding {} publishers on topic {}", arguments.numProducers, topic);
for (int j = 0; j < arguments.numProducers; j++) {
futures.add(producerBuilder.clone().topic(topic).createAsync());
}
}
final List<Producer<byte[]>> producers = Lists.newArrayListWithCapacity(futures.size());
for (Future<Producer<byte[]>> future : futures) {
producers.add(future.get());
}
log.info("Created {} producers", producers.size());
long start = System.nanoTime();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
......@@ -341,81 +295,26 @@ public class PerformanceProducer {
printAggregatedStats();
}));
Collections.shuffle(producers);
AtomicBoolean isDone = new AtomicBoolean();
executor.submit(() -> {
try {
RateLimiter rateLimiter = RateLimiter.create(arguments.msgRate);
long startTime = System.nanoTime();
long warmupEndTime = startTime + (long) (arguments.warmupTimeSeconds * 1e9);
long testEndTime = startTime + (long) (arguments.testTime * 1e9);
// Send messages on all topics/producers
long totalSent = 0;
while (true) {
for (Producer<byte[]> producer : producers) {
if (arguments.testTime > 0) {
if (System.nanoTime() > testEndTime) {
log.info("------------------- DONE -----------------------");
printAggregatedStats();
isDone.set(true);
Thread.sleep(5000);
System.exit(0);
}
}
if (arguments.numMessages > 0) {
if (totalSent++ >= arguments.numMessages) {
log.info("------------------- DONE -----------------------");
printAggregatedStats();
isDone.set(true);
Thread.sleep(5000);
System.exit(0);
}
}
rateLimiter.acquire();
final long sendTime = System.nanoTime();
byte[] payloadData;
if (arguments.payloadFilename != null) {
payloadData = payloadByteList.get(random.nextInt(payloadByteList.size()));
} else {
payloadData = payloadBytes;
}
TypedMessageBuilder<byte[]> messageBuilder = producer.newMessage()
.value(payloadData);
if (arguments.delay >0) {
messageBuilder.deliverAfter(arguments.delay, TimeUnit.SECONDS);
}
messageBuilder.sendAsync().thenRun(() -> {
messagesSent.increment();
bytesSent.add(payloadData.length);
totalMessagesSent.increment();
totalBytesSent.add(payloadData.length);
long now = System.nanoTime();
if (now > warmupEndTime) {
long latencyMicros = NANOSECONDS.toMicros(now - sendTime);
recorder.recordValue(latencyMicros);
cumulativeRecorder.recordValue(latencyMicros);
}
}).exceptionally(ex -> {
log.warn("Write error on message", ex);
System.exit(-1);
return null;
});
}
}
} catch (Throwable t) {
log.error("Got error", t);
}
});
CountDownLatch doneLatch = new CountDownLatch(arguments.numTestThreads);
final long numMessagesPerThread = arguments.numMessages / arguments.numTestThreads;
final int msgRatePerThread = arguments.msgRate / arguments.numTestThreads;
for (int i = 0; i < arguments.numTestThreads; i++) {
final int threadIdx = i;
executor.submit(() -> {
log.info("Started performance test thread {}", threadIdx);
runProducer(
arguments,
numMessagesPerThread,
msgRatePerThread,
payloadByteList,
payloadBytes,
random,
doneLatch
);
});
}
// Print report stats
long oldTime = System.nanoTime();
......@@ -439,7 +338,7 @@ public class PerformanceProducer {
break;
}
if (isDone.get()) {
if (doneLatch.getCount() <= 0) {
break;
}
......@@ -467,8 +366,153 @@ public class PerformanceProducer {
oldTime = now;
}
}
private static void runProducer(Arguments arguments,
long numMessages,
int msgRate,
List<byte[]> payloadByteList,
byte[] payloadBytes,
Random random,
CountDownLatch doneLatch) {
PulsarClient client = null;
try {
// Now processing command line arguments
String prefixTopicName = arguments.topics.get(0);
List<Future<Producer<byte[]>>> futures = Lists.newArrayList();
ClientBuilder clientBuilder = PulsarClient.builder() //
.serviceUrl(arguments.serviceURL) //
.connectionsPerBroker(arguments.maxConnections) //
.ioThreads(Runtime.getRuntime().availableProcessors()) //
.statsInterval(arguments.statsIntervalSeconds, TimeUnit.SECONDS) //
.tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);
if (isNotBlank(arguments.authPluginClassName)) {
clientBuilder.authentication(arguments.authPluginClassName, arguments.authParams);
}
client = clientBuilder.build();
ProducerBuilder<byte[]> producerBuilder = client.newProducer() //
.sendTimeout(0, TimeUnit.SECONDS) //
.compressionType(arguments.compression) //
.maxPendingMessages(arguments.maxOutstanding) //
.maxPendingMessagesAcrossPartitions(arguments.maxPendingMessagesAcrossPartitions)
// enable round robin message routing if it is a partitioned topic
.messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
if (arguments.batchTimeMillis == 0.0) {
producerBuilder.enableBatching(false);
} else {
long batchTimeUsec = (long) (arguments.batchTimeMillis * 1000);
producerBuilder.batchingMaxPublishDelay(batchTimeUsec, TimeUnit.MICROSECONDS)
.enableBatching(true);
}
// Block if queue is full else we will start seeing errors in sendAsync
producerBuilder.blockIfQueueFull(true);
if (arguments.encKeyName != null) {
producerBuilder.addEncryptionKey(arguments.encKeyName);
byte[] pKey = Files.readAllBytes(Paths.get(arguments.encKeyFile));
EncKeyReader keyReader = new EncKeyReader(arguments.encKeyName, pKey);
producerBuilder.cryptoKeyReader(keyReader);
}
for (int i = 0; i < arguments.numTopics; i++) {
String topic = (arguments.numTopics == 1) ? prefixTopicName : String.format("%s-%d", prefixTopicName, i);
log.info("Adding {} publishers on topic {}", arguments.numProducers, topic);
client.close();
for (int j = 0; j < arguments.numProducers; j++) {
futures.add(producerBuilder.clone().topic(topic).createAsync());
}
}
final List<Producer<byte[]>> producers = Lists.newArrayListWithCapacity(futures.size());
for (Future<Producer<byte[]>> future : futures) {
producers.add(future.get());
}
Collections.shuffle(producers);
log.info("Created {} producers", producers.size());
RateLimiter rateLimiter = RateLimiter.create(msgRate);
long startTime = System.nanoTime();
long warmupEndTime = startTime + (long) (arguments.warmupTimeSeconds * 1e9);
long testEndTime = startTime + (long) (arguments.testTime * 1e9);
// Send messages on all topics/producers
long totalSent = 0;
while (true) {
for (Producer<byte[]> producer : producers) {
if (arguments.testTime > 0) {
if (System.nanoTime() > testEndTime) {
log.info("------------------- DONE -----------------------");
printAggregatedStats();
doneLatch.countDown();
Thread.sleep(5000);
System.exit(0);
}
}
if (numMessages > 0) {
if (totalSent++ >= numMessages) {
log.info("------------------- DONE -----------------------");
printAggregatedStats();
doneLatch.countDown();
Thread.sleep(5000);
System.exit(0);
}
}
rateLimiter.acquire();
final long sendTime = System.nanoTime();
byte[] payloadData;
if (arguments.payloadFilename != null) {
payloadData = payloadByteList.get(random.nextInt(payloadByteList.size()));
} else {
payloadData = payloadBytes;
}
TypedMessageBuilder<byte[]> messageBuilder = producer.newMessage()
.value(payloadData);
if (arguments.delay >0) {
messageBuilder.deliverAfter(arguments.delay, TimeUnit.SECONDS);
}
messageBuilder.sendAsync().thenRun(() -> {
messagesSent.increment();
bytesSent.add(payloadData.length);
totalMessagesSent.increment();
totalBytesSent.add(payloadData.length);
long now = System.nanoTime();
if (now > warmupEndTime) {
long latencyMicros = NANOSECONDS.toMicros(now - sendTime);
recorder.recordValue(latencyMicros);
cumulativeRecorder.recordValue(latencyMicros);
}
}).exceptionally(ex -> {
log.warn("Write error on message", ex);
System.exit(-1);
return null;
});
}
}
} catch (Throwable t) {
log.error("Got error", t);
} finally {
if (null != client) {
try {
client.close();
} catch (PulsarClientException e) {
log.error("Failed to close test client", e);
}
}
}
}
private static void printAggregatedThroughput(long start) {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或