在Apache Pulsar中创建分区主题的策略与实践

# 允许自动创建主题的类型：非分区 (non-partitioned) 或 分区 (partitioned)
# 设置为 partitioned 后，客户端首次访问不存在的主题时，将自动创建分区主题。
allowAutoTopicCreationType = partitioned

# 当 allowAutoTopicCreationType 设置为 partitioned 时，自动创建的分区主题的默认分区数量
# 请根据预期的吞吐量和消费者并行度合理设置此值。
defaultNumPartitions = <N>

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;

public class AutoPartitionedTopicExample {

    public Producer<byte[]> createProducer(PulsarClient pulsarClient, String topicName) throws PulsarClientException {
        // 如果 Broker 配置了 allowAutoTopicCreationType = partitioned 和 defaultNumPartitions = 3
        // 并且 topicName ("my-auto-topic") 之前不存在，Pulsar 将自动创建一个名为 "my-auto-topic" 的3分区主题。
        return pulsarClient.newProducer()
                .topic(topicName)
                .create();
    }

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650") // 替换为你的Pulsar服务URL
                .build();

        AutoPartitionedTopicExample example = new AutoPartitionedTopicExample();
        Producer<byte[]> producer = example.createProducer(client, "persistent://public/default/my-auto-topic");
        System.out.println("Producer created for topic: " + producer.getTopic());
        producer.close();
        client.close();
    }
}

import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.PulsarClientException;

public class PulsarAdminTopicCreator {

    public static void main(String[] args) throws PulsarClientException, PulsarAdminException {
        String adminServiceUrl = "http://localhost:8080"; // 替换为你的Pulsar Admin服务URL
        String tenant = "public";
        String namespace = "default";
        String topicName = "my-manual-partitioned-topic"; // 主题的短名称
        int numPartitions = 5; // 期望的分区数量

        // 完整的Topic名称格式：persistent://tenant/namespace/topicName
        String fullTopicName = String.format("persistent://%s/%s/%s", tenant, namespace, topicName);

        try (PulsarAdmin pulsarAdmin = PulsarAdmin.builder()
                .serviceHttpUrl(adminServiceUrl)
                .build()) {

            // 检查主题是否已存在，避免重复创建
            // 注意：getPartitionedTopicList 返回的是完整主题名列表
            if (!pulsarAdmin.topics().getPartitionedTopicList(tenant + "/" + namespace).contains(fullTopicName)) {
                System.out.println("Creating partitioned topic: " + fullTopicName + " with " + numPartitions + " partitions.");
                pulsarAdmin.topics().createPartitionedTopic(fullTopicName, numPartitions);
                System.out.println("Partitioned topic created successfully.");
            } else {
                System.out.println("Partitioned topic " + fullTopicName + " already exists.");
                // 如果需要，可以获取现有分区数量进行验证
                // int existingPartitions = pulsarAdmin.topics().getPartitionedTopicMetadata(fullTopicName).partitions;
                // System.out.println("Existing partitions: " + existingPartitions);
            }

        } catch (PulsarAdminException e) {
            System.err.println("Failed to create partitioned topic: " + e.getMessage());
            throw e;
        }
    }
}