Apache Pulsar 主题分区创建与管理指南

# 允许自动创建的主题类型。
# 可选值包括：non-partitioned (非分区), partitioned (分区), none (不允许自动创建)
allowAutoTopicCreationType = partitioned

# 当 allowAutoTopicCreationType 设置为 partitioned 时，
# 自动创建主题的默认分区数量
defaultNumPartitions = 5

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

public class PartitionedTopicCreator {

    private static final String SERVICE_URL = "http://localhost:8080"; // Pulsar Admin API 的服务URL
    private static final String TOPIC_NAME = "persistent://public/default/my-partitioned-topic";
    private static final int NUM_PARTITIONS = 3; // 期望的分区数量

    public static void main(String[] args) {
        PulsarAdmin pulsarAdmin = null;
        try {
            // 1. 构建 PulsarAdmin 实例
            pulsarAdmin = PulsarAdmin.builder()
                    .serviceUrl(SERVICE_URL)
                    .build();

            // 2. 检查主题是否已存在，避免重复创建
            if (pulsarAdmin.topics().getPartitionedTopicMetadata(TOPIC_NAME).partitions == 0) {
                System.out.println("主题 " + TOPIC_NAME + " 不存在或不是分区主题，准备创建...");
                // 3. 使用 createPartitionedTopic 方法创建分区主题
                pulsarAdmin.topics().createPartitionedTopic(TOPIC_NAME, NUM_PARTITIONS);
                System.out.println("成功创建分区主题: " + TOPIC_NAME + "，分区数量: " + NUM_PARTITIONS);
            } else {
                System.out.println("分区主题 " + TOPIC_NAME + " 已存在，分区数量: " + 
                                   pulsarAdmin.topics().getPartitionedTopicMetadata(TOPIC_NAME).partitions);
            }

        } catch (PulsarClientException e) {
            System.err.println("Pulsar 客户端异常: " + e.getMessage());
            e.printStackTrace();
        } catch (PulsarAdminException e) {
            if (e.getStatusCode() == 409) { // 409 Conflict 表示资源已存在
                System.out.println("分区主题 " + TOPIC_NAME + " 已存在。");
            } else {
                System.err.println("Pulsar Admin 异常: " + e.getMessage());
                e.printStackTrace();
            }
        } finally {
            if (pulsarAdmin != null) {
                try {
                    pulsarAdmin.close();
                } catch (PulsarAdminException e) {
                    System.err.println("关闭 PulsarAdmin 异常: " + e.getMessage());
                }
            }
        }
    }
}

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class PulsarProducerConfig {

    private final String testTopicName = "persistent://public/default/test"; // 假设这是已创建的分区主题

    @Bean(name = "testProducer")
    public Producer<TestMessage> testProducer(PulsarClient pulsarClient) throws PulsarClientException {
        return pulsarClient.newProducer()
                .topic(testTopicName)
                // 其他可选配置，如消息路由模式、批处理等
                // .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                // .enableBatching(true)
                .create();
    }

    // 假设 TestMessage 是您的消息类型
    public static class TestMessage {
        private String content;
        // constructor, getters, setters
    }
}