理解 Apache Pulsar 分区主题
在 Apache Pulsar 中，主题（Topic）是消息生产和消费的核心。为了提高吞吐量、实现并行处理和负载均衡，Pulsar 引入了分区主题（Partitioned Topic）的概念。一个分区主题由多个独立的分区组成，每个分区都可以被视为一个独立的、有序的日志流。生产者可以将消息发送到特定分区，消费者可以从一个或多个分区消费消息，从而实现更高的并发度。
当在 Java 应用中创建 Pulsar 生产者（Producer）并指定一个主题时，如果该主题尚未存在，Pulsar 会根据其配置自动创建它。然而，默认情况下创建的主题是非分区的。要确保主题在创建时就是分区主题，需要采取特定的策略。
创建分区主题的两种策略
在 Apache Pulsar 中，主要有两种方法来创建分区主题，这两种方法都要求在主题首次创建时就指定分区属性。
1. 通过 Broker 配置实现自动分区创建
如果您的 Pulsar 环境允许自动创建主题，并且您希望所有自动创建的主题都是分区主题，可以通过修改 Pulsar Broker 的配置文件（通常是 broker.conf）来实现。这是一种全局性的配置，适用于所有通过自动创建机制生成的新主题。
立即学习“Java免费学习笔记（深入）”；
要启用此功能，请在 broker.conf 中设置以下参数：
# 允许自动创建分区主题的类型，设置为 'partitioned' allowAutoTopicCreationType = partitioned # 默认的分区数量，例如设置为 5 defaultNumPartitions = 5
其中，defaultNumPartitions 是您希望为自动创建的分区主题设置的默认分区数量。
注意事项：
- 这种方法适用于那些希望所有新主题都具有相同默认分区数量的场景。
- 一旦 Broker 配置更改并重启，后续通过生产者首次访问而自动创建的主题都将是分区主题。
- 对于需要不同分区数量的主题，或者不希望所有自动创建主题都是分区主题的场景，此方法可能不适用。
2. 使用 Pulsar Admin API 显式创建分区主题
对于需要更精细控制分区数量，或者不希望依赖 Broker 默认配置的场景，推荐使用 Pulsar Admin API 显式地创建分区主题。这种方法允许您在应用程序或管理脚本中，以编程方式指定主题的分区数量。
以下是一个使用 Java Pulsar Admin API 创建分区主题的示例：
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.PulsarClientException;
public class PartitionedTopicCreator {
// Pulsar Admin Service URL，通常是 Pulsar Broker 的 HTTP 端口
private static final String SERVICE_URL = "http://localhost:8080";
public static void main(String[] args) {
String tenant = "public";
String namespace = "default";
// 完整的 Pulsar 主题名称格式：persistent://租户/命名空间/主题名
String topicName = "persistent://" + tenant + "/" + namespace + "/my-partitioned-topic";
int numPartitions = 3; // 指定分区数量
PulsarAdmin pulsarAdmin = null;
try {
pulsarAdmin = PulsarAdmin.builder()
.serviceHttpUrl(SERVICE_URL)
.build();
// 尝试获取分区主题的元数据，如果主题不存在或不是分区主题会抛出异常
try {
int existingPartitions = pulsarAdmin.topics().getPartitionedTopicMetadata(topicName).getPartitions().size();
System.out.println("Topic '" + topicName + "' already exists with " + existingPartitions + " partitions.");
} catch (PulsarAdminException.NotFoundException e) {
// 主题不存在，可以创建
pulsarAdmin.topics().createPartitionedTopic(topicName, numPartitions);
System.out.println("Partitioned topic '" + topicName + "' created with " + numPartitions + " partitions.");
}
} catch (PulsarClientException e) {
System.err.println("Failed to connect to Pulsar Admin: " + e.getMessage());
} catch (PulsarAdminException e) {
System.err.println("Failed to manage partitioned topic: " + e.getMessage());
} finally {
if (pulsarAdmin != null) {
try {
pulsarAdmin.close();
} catch (PulsarAdminException e) {
System.err.println("Error closing PulsarAdmin: " + e.getMessage());
}
}
}
}
}在上述代码中，pulsarAdmin.topics().createPartitionedTopic(topicName, numPartitions) 方法是核心，它负责向 Pulsar 集群发送请求，创建一个指定分区数量的主题。
关键注意事项：分区主题的不可变性
一旦主题被创建，无论是作为非分区主题还是分区主题，其分区属性都是不可更改的。 这意味着，一个已存在的非分区主题无法通过任何方式（包括 Admin API）转换为分区主题，也无法更改一个已存在分区主题的分区数量。尝试对已存在的非分区主题进行分区操作，或更改已分区主题的分区数量，都将失败。
因此，在设计和部署 Pulsar 应用时，务必提前规划好主题的分区策略。如果需要一个分区主题，必须在主题首次创建时就将其创建为分区主题。如果错误地创建了非分区主题，唯一的解决方案是删除该主题（如果允许数据丢失）并重新创建一个分区主题，或者创建一个新的分区主题并迁移数据。
在 Java 应用中集成分区主题
当您使用 @Bean 注解在 Spring 应用程序中定义 Pulsar 生产者时，您所引用的主题必须在此之前已经被创建为分区主题。您的 @Bean 定义仅仅是创建了一个生产者实例，并将其绑定到一个已存在的主题上。Pulsar 客户端在连接到主题时，会根据主题的实际类型（分区或非分区）进行相应的操作。
考虑以下示例，一个名为 test 的分区主题已经通过上述任一策略创建完成：
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class PulsarProducerConfig {
// 假设 "test" 主题已经通过 Admin API 或 Broker 配置被创建为分区主题
private static final String TEST_TOPIC_NAME = "persistent://public/default/test";
/**
* 定义一个 Pulsar 客户端 Bean
*/
@Bean
public PulsarClient pulsarClient() throws PulsarClientException {
return PulsarClient.builder()
.serviceUrl("pulsar://localhost:6650") // Pulsar Broker Service URL
.build();
}
/**
* 定义一个 Pulsar 生产者 Bean
* 确保 TEST_TOPIC_NAME 对应的 Pulsar 主题已是分区主题
*/
@Bean(name = "testProducer")
public Producer testProducer(PulsarClient pulsarClient) throws PulsarClientException {
return pulsarClient.newProducer()
.topic(TEST_TOPIC_NAME)
.producerName("my-test-producer") // 可选：指定生产者名称
.create();
}
} 在这个场景中，testProducer bean 只是创建了一个指向 TEST_TOPIC_NAME 的生产者。如果 TEST_TOPIC_NAME 在此之前不是分区主题，那么这个生产者将与一个非分区主题交互，无法获得分区带来的优势。因此，确保主题在生产者创建之前就已正确分区是关键。
总结
创建 Apache Pulsar 分区主题是实现高吞吐量和可扩展性消息系统的基础。本文介绍了两种主要的创建策略：通过 Pulsar Broker 配置实现自动分区创建，以及通过 Pulsar Admin API 显式创建。无论采用哪种方法，核心原则是分区主题必须在创建时就指定分区数量，且一旦创建，其分区属性便不可更改。 在 Java 应用中集成生产者时，务必确保目标主题已按照预期被正确分区，以充分利用 Pulsar 的并行处理能力。
