public static void main(String... args) throws Exception {

// TODO(developer): Replace these variables before running the sample.

String projectId = "your-project-id";

String topicId = "your-topic-id";

publishWithConcurrencyControlExample(projectId, topicId);

}

public static void publishWithConcurrencyControlExample(String projectId, String topicId)

throws IOException, ExecutionException, InterruptedException {

TopicName topicName = TopicName.of(projectId, topicId);

Publisher publisher = null;

List<ApiFuture<String>> messageIdFutures = new ArrayList<>();

try {

// Provides an executor service for processing messages. The default

// `executorProvider` used by the publisher has a default thread count of

// 5 * the number of processors available to the Java virtual machine.

ExecutorProvider executorProvider =

InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(4).build();

// `setExecutorProvider` configures an executor for the publisher.

publisher = Publisher.newBuilder(topicName).setExecutorProvider(executorProvider).build();

// schedule publishing one message at a time : messages get automatically batched

for (int i = 0; i < 100; i++) {

String message = "message " + i;

ByteString data = ByteString.copyFromUtf8(message);

PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

// Once published, returns a server-assigned message id (unique within the topic)

ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);

messageIdFutures.add(messageIdFuture);

}

} finally {

// Wait on any pending publish requests.

List<String> messageIds = ApiFutures.allAsList(messageIdFutures).get();

System.out.println("Published " + messageIds.size() + " messages with concurrency control.");

if (publisher != null) {

// When finished with the publisher, shutdown to free up resources.

publisher.shutdown();

publisher.awaitTermination(1, TimeUnit.MINUTES);

}

}

}