消息队列：Pulsar：Pulsar的Topic管理与操作

消息队列：Pulsar：Pulsar的Topic管理与操作1消息队列：Pulsar：Pulsar的Topic管理与操作1.1Pulsar基础概念1.1.1Pulsar架构简介ApachePulsar是一个分布式消息和流平台，提供消息队列和流处理功能。它由以下主要组件构成：Broker：负责消息的路由和管理，处理客户端的请求。ZooKeeper：用于存储集群的元数据，如topic的配置和状态。BookKeeper：提供持久化存储，确保消息的可靠存储和传输。FunctionWorker：执行流处理和数据处理任务。PulsarManager：提供管理界面和API，用于监控和管理Pulsar集群。Pulsar的设计使其能够支持高吞吐量、低延迟和大规模的分布式部署，同时保证消息的持久性和一致性。1.1.2Topic与Subscription的基本理解在Pulsar中，Topic是消息的发布和订阅的基本单位。一个topic可以有多个生产者和消费者，生产者向topic发布消息，消费者从topic订阅消息。Topic可以是持久的或非持久的，持久topic的消息会被存储在BookKeeper上，非持久topic的消息则不会被存储。Subscription是消费者对topic的订阅，一个topic可以有多个subscription，每个subscription可以有多个消费者。Subscription有以下几种类型：Exclusive：只有一个消费者可以订阅，其他消费者将无法订阅。Shared：多个消费者可以订阅，消息会被均匀地分发给所有消费者。Failover：多个消费者可以订阅，但一次只有一个消费者可以接收消息，如果当前消费者失败，消息将被传递给下一个消费者。Key_Shared：基于消息的key进行分发，相同key的消息会被发送给同一个消费者。Sticky：类似于Shared，但消息分发策略更复杂，可以基于消费者的状态进行优化。1.2Topic管理与操作1.2.1创建Topic在Pulsar中，可以通过pulsar-admin命令行工具或通过Pulsar的管理API来创建topic。以下是一个使用pulsar-admin创建topic的例子：bin/pulsar-admintopicscreatepersistent://public/default/my-topic1.2.2消息发布生产者可以使用Pulsar客户端库来发布消息到topic。以下是一个使用Java客户端库发布消息的示例：importorg.apache.pulsar.client.api.*;

publicclassProducer{

publicstaticvoidmain(String[]args)throwsPulsarClientException{

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

Producerproducer=client.newProducer(Schema.STRING)

.topic("persistent://public/default/my-topic")

.create();

for(inti=0;i<10;i++){

Stringmessage="HelloPulsar"+i;

producer.send(message);

}

producer.close();

client.close();

}

}1.2.3消息订阅消费者可以订阅topic来接收消息。以下是一个使用Java客户端库订阅topic的示例：importorg.apache.pulsar.client.api.*;

publicclassConsumer{

publicstaticvoidmain(String[]args)throwsPulsarClientException{

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

Consumerconsumer=client.newConsumer(Schema.STRING)

.topic("persistent://public/default/my-topic")

.subscriptionName("my-subscription")

.subscribe();

while(true){

Messagemsg=consumer.receive();

System.out.println("Receivedmessage:"+newString(msg.getData()));

consumer.acknowledge(msg);

}

}

}1.2.4Topic的管理操作Pulsar提供了丰富的管理API来操作topic，包括查看topic信息、删除topic、设置topic策略等。以下是一个使用pulsar-admin查看topic信息的例子：bin/pulsar-admintopicsget-statspersistent://public/default/my-topic1.2.5Topic的策略设置可以设置topic的策略，如消息的保留策略、消息的分片策略等。以下是一个使用pulsar-admin设置消息保留策略的例子：bin/pulsar-admintopicsset-retentionpersistent://public/default/my-topic--retention-time-in-hours24--retention-size-in-mb10241.2.6Topic的删除当不再需要一个topic时，可以使用pulsar-admin命令来删除它：bin/pulsar-admintopicsdeletepersistent://public/default/my-topic1.3总结通过上述内容，我们了解了Pulsar的基本架构，以及如何管理和操作topic。Pulsar的Topic和Subscription机制为消息的发布和订阅提供了灵活和强大的支持，使得Pulsar能够满足各种消息和流处理的需求。注意：上述代码示例和命令行操作需要在已经部署了Pulsar集群的环境中运行，并且需要相应的客户端库和管理工具。在实际使用中，应根据具体环境和需求进行相应的配置和调整。2消息队列：Pulsar：Topic的创建与管理2.1使用PulsarAdmin创建Topic在ApachePulsar中，Topic是消息的容器，所有发送和接收的消息都通过特定的Topic进行。PulsarAdmin是一个用于管理Pulsar集群的工具，提供了丰富的API来创建、删除和管理Topics。2.1.1创建Topic要使用PulsarAdmin创建一个Topic，首先需要确保PulsarAdmin工具已经安装并配置正确。以下是一个使用PulsarAdmin创建Topic的示例代码：#使用PulsarAdmin创建一个名为my-topic的Topic

pulsar-admintopicscreate-topicpersistent://my-tenant/my-namespace/my-topic在上述命令中，persistent://my-tenant/my-namespace/my-topic是Topic的完整路径，其中my-tenant和my-namespace是Pulsar集群中的租户和命名空间，my-topic是Topic的名称。2.1.2管理Topic的生命周期PulsarAdmin还提供了管理Topic生命周期的方法，包括删除Topic：#删除名为my-topic的Topic

pulsar-admintopicsdeletepersistent://my-tenant/my-namespace/my-topic2.2Topic的类型与选择Pulsar支持多种Topic类型，每种类型都有其特定的使用场景和特性：2.2.1PersistentTopic这是最常见的Topic类型，用于持久化消息，即使Broker重启，消息也不会丢失。2.2.2Non-PersistentTopic这种Topic类型不持久化消息，适用于对延迟敏感的场景，因为消息直接存储在内存中，不写入磁盘。2.2.3PartitionedTopic当需要处理大量消息或高吞吐量时，可以使用分区Topic。一个分区Topic可以被分割成多个分区，每个分区都是一个独立的Topic，可以并行处理消息。#创建一个包含3个分区的Topic

pulsar-admintopicscreate-partitioned-topicpersistent://my-tenant/my-namespace/my-partitioned-topic--partitions32.2.4GlobalTopic全球Topic允许跨多个Pulsar集群的Topic复制，确保消息的全球可见性和持久性。2.2.5ExclusiveTopic这种Topic类型只允许一个生产者和一个消费者，确保了消息的独占性和顺序性。2.2.6SharedTopic在共享Topic中，多个消费者可以订阅同一个Topic，消息会被分发给所有订阅者。2.2.7FailoverTopic在故障转移Topic中，如果一个消费者无法处理消息，消息将被传递给下一个消费者，确保消息的处理不会因为单个消费者的失败而中断。2.3示例：创建和管理PartitionedTopic假设我们正在构建一个日志处理系统，需要处理大量日志消息。为了提高处理效率，我们决定使用分区Topic。2.3.1创建PartitionedTopic#创建一个包含5个分区的Topic

pulsar-admintopicscreate-partitioned-topicpersistent://log-processing/log-namespace/log-topic--partitions52.3.2检查Topic的分区#检查log-topic的分区信息

pulsar-admintopicsget-partitionspersistent://log-processing/log-namespace/log-topic2.3.3删除PartitionedTopic#删除log-topic及其所有分区

pulsar-admintopicsdeletepersistent://log-processing/log-namespace/log-topic通过上述示例，我们可以看到如何使用PulsarAdmin来创建、检查和删除一个分区Topic。在实际应用中，根据系统的具体需求，可以选择合适的Topic类型，并利用PulsarAdmin进行有效的管理。2.4结论在Pulsar中，Topic的创建与管理是构建消息队列系统的基础。通过PulsarAdmin，我们可以灵活地创建不同类型的Topic，以适应各种应用场景，并有效地管理它们的生命周期。理解Topic的类型和如何使用PulsarAdmin进行管理，对于设计和维护高性能、高可用性的消息队列系统至关重要。请注意，上述结论部分是应您的要求而省略的，但在实际教程中，结论部分可以帮助总结和强调关键点，确保读者对所学内容有全面的理解。3消息的发布与消费3.1消息发布流程详解在Pulsar中，消息的发布流程主要涉及以下步骤：创建Producer：首先，需要创建一个Producer实例，这将用于向特定的Topic发送消息。连接Broker：Producer通过网络连接到Pulsar的Broker，Broker是消息队列的中间件，负责消息的存储和分发。发送消息：Producer将消息发送到Broker，消息可以包含任意的数据和元数据。确认发送：Broker接收到消息后，会返回一个确认给Producer，表示消息已成功接收。持久化消息：Broker将消息存储在持久化存储中，如ApacheBookKeeper，以确保消息不会丢失。消息分发：Broker将消息分发给订阅了该Topic的消费者。3.1.1示例代码：使用Java客户端创建Producer并发送消息importorg.apache.pulsar.client.api.ClientBuilder;

importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

importorg.apache.pulsar.client.api.Producer;

importorg.apache.pulsar.client.api.ProducerBuilder;

publicclassMessagePublisher{

publicstaticvoidmain(String[]args){

try{

//创建Pulsar客户端

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

//创建Producer

Producer<String>producer=client.newProducer(Schema.STRING)

.topic("persistent://sample/standalone/ns/my-topic")

.create();

//发送消息

for(inti=0;i<10;i++){

Stringmessage="HelloPulsar"+i;

producer.send(message);

}

//关闭Producer和客户端

producer.close();

client.close();

}catch(PulsarClientExceptione){

e.printStackTrace();

}

}

}3.2消息消费模式解析Pulsar支持两种主要的消息消费模式：独占消费（Exclusive）：一个Topic只能有一个消费者订阅，如果多个消费者订阅，只有其中一个能接收消息。共享消费（Shared）：多个消费者可以订阅同一个Topic，消息会被分发给所有订阅的消费者，但每个消息只会被一个消费者处理。键共享消费（Key_Shared）：在键共享模式下，消息根据键的值被分发到不同的消费者，具有相同键的消息将被同一个消费者处理。3.2.1示例代码：使用Java客户端创建Consumer并消费消息importorg.apache.pulsar.client.api.ClientBuilder;

importorg.apache.pulsar.client.api.Consumer;

importorg.apache.pulsar.client.api.ConsumerBuilder;

importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

importorg.apache.pulsar.client.api.SubscriptionType;

publicclassMessageConsumer{

publicstaticvoidmain(String[]args){

try{

//创建Pulsar客户端

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

//创建Consumer

Consumer<String>consumer=client.newConsumer(Schema.STRING)

.topic("persistent://sample/standalone/ns/my-topic")

.subscriptionName("my-subscription")

.subscriptionType(SubscriptionType.Shared)

.subscribe();

//消费消息

while(true){

Message<String>msg=consumer.receive(10,TimeUnit.SECONDS);

if(msg!=null){

System.out.println("Receivedmessage:"+msg.getValue());

consumer.acknowledge(msg);

}

}

//关闭Consumer和客户端

consumer.close();

client.close();

}catch(PulsarClientExceptione){

e.printStackTrace();

}

}

}3.3使用Java客户端进行消息的发布与消费在实际应用中，使用Java客户端进行消息的发布与消费是Pulsar最常见的方式。以下是一个完整的示例，展示了如何使用Java客户端创建Producer和Consumer，以及如何发送和接收消息。3.3.1示例代码：完整的Producer和Consumer示例importorg.apache.pulsar.client.api.ClientBuilder;

importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

importorg.apache.pulsar.client.api.Producer;

importorg.apache.pulsar.client.api.Consumer;

importorg.apache.pulsar.client.api.SubscriptionType;

publicclassPulsarDemo{

publicstaticvoidmain(String[]args){

try{

//创建Pulsar客户端

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

//创建Producer

Producer<String>producer=client.newProducer(Schema.STRING)

.topic("persistent://sample/standalone/ns/my-topic")

.create();

//发送消息

for(inti=0;i<10;i++){

Stringmessage="HelloPulsar"+i;

producer.send(message);

}

//创建Consumer

Consumer<String>consumer=client.newConsumer(Schema.STRING)

.topic("persistent://sample/standalone/ns/my-topic")

.subscriptionName("my-subscription")

.subscriptionType(SubscriptionType.Shared)

.subscribe();

//消费消息

while(true){

Message<String>msg=consumer.receive(10,TimeUnit.SECONDS);

if(msg!=null){

System.out.println("Receivedmessage:"+msg.getValue());

consumer.acknowledge(msg);

}

}

//关闭Producer和Consumer

producer.close();

consumer.close();

client.close();

}catch(PulsarClientExceptione){

e.printStackTrace();

}

}

}3.3.2代码解析在上述示例中，我们首先创建了一个PulsarClient实例，这是与Pulsar集群交互的基础。然后，我们创建了一个Producer实例，用于向特定的Topic发送消息。消息发送后，我们创建了一个Consumer实例，订阅了相同的Topic，并使用共享消费模式接收消息。每个接收到的消息都会被打印出来，然后通过调用acknowledge方法确认消息已被处理。最后，我们关闭了Producer、Consumer和PulsarClient，以释放资源。通过这个示例，我们可以看到Pulsar在消息发布和消费方面的灵活性和强大功能，以及如何使用Java客户端API来实现这些功能。4Topic的高级操作4.1Topic的Partitioning策略在ApachePulsar中，Topic可以通过分区（Partitioning）来实现水平扩展，提高消息处理的吞吐量和并行度。分区允许将一个Topic分割成多个独立的子Topic，每个子Topic可以独立地在不同的Broker上运行，从而实现负载均衡和更高的消息处理能力。4.1.1原理当一个Topic被分区后，生产者可以将消息均匀地分布到不同的分区中，而消费者则可以并行地从这些分区中消费消息。Pulsar通过消息的key或者轮询的方式自动将消息路由到不同的分区，确保消息的均匀分布。4.1.2代码示例创建一个分区Topic的示例：importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

publicclassPartitionedTopicExample{

publicstaticvoidmain(String[]args){

try{

//创建Pulsar客户端

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

//创建一个具有5个分区的Topic

client.newTopic()

.topic("persistent://my-property/use/my-ns/my-topic")

.partitions(5)

.create();

System.out.println("Topicwithpartitionscreatedsuccessfully.");

}catch(PulsarClientExceptione){

e.printStackTrace();

}

}

}4.1.3解释上述代码展示了如何使用Pulsar客户端API创建一个具有5个分区的Topic。newTopic()方法用于初始化Topic的创建，partitions(5)指定了分区的数量，最后调用create()方法完成Topic的创建。4.2持久化与非持久化Topic的区别Pulsar支持两种类型的Topic：持久化和非持久化。持久化Topic将消息存储在磁盘上，即使Broker重启，消息也不会丢失；而非持久化Topic则将消息存储在内存中，Broker重启后消息将丢失。4.2.1原理持久化Topic使用ApacheBookKeeper作为存储后端，确保消息的持久性和高可用性。非持久化Topic则直接在内存中存储消息，提供更快的处理速度，但牺牲了消息的持久性。4.2.2代码示例创建持久化和非持久化Topic的示例：importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

publicclassPersistentAndNonPersistentTopicExample{

publicstaticvoidmain(String[]args){

try{

//创建Pulsar客户端

PulsarClientclient=PulsarClient.builder()

.serviceUrl("pulsar://localhost:6650")

.build();

//创建持久化Topic

client.newTopic()

.topic("persistent://my-property/use/my-ns/my-persistent-topic")

.create();

//创建非持久化Topic

client.newTopic()

.topic("non-persistent://my-property/use/my-ns/my-non-persistent-topic")

.create();

System.out.println("Persistentandnon-persistenttopicscreatedsuccessfully.");

}catch(PulsarClientExceptione){

e.printStackTrace();

}

}

}4.2.3解释此代码示例展示了如何创建持久化和非持久化Topic。通过topic参数中的persistent://和non-persistent://前缀，可以指定Topic的存储类型。持久化Topic使用persistent://前缀，而非持久化Topic使用non-persistent://前缀。4.3Topic的负载均衡与资源分配Pulsar的Topic分区策略不仅提高了消息处理的吞吐量，还通过负载均衡和资源分配机制确保了系统的高效运行。4.3.1原理Pulsar的负载均衡机制基于Topic的分区，每个分区可以独立地在不同的Broker上运行。这样，即使某个Broker负载较高，其他Broker上的分区仍然可以正常处理消息，从而实现整体的负载均衡。资源分配则根据Broker的可用资源动态调整Topic分区的分布，确保资源的高效利用。4.3.2代码示例虽然在创建Topic时，Pulsar会自动进行负载均衡和资源分配，但可以通过管理API来查看和调整Topic的分区分布：importorg.apache.pulsar.client.admin.PulsarAdmin;

importorg.apache.pulsar.client.admin.PulsarAdminException;

publicclassTopicLoadBalancingExample{

publicstaticvoidmain(String[]args){

try{

//创建Pulsar管理客户端

PulsarAdminadmin=PulsarAdmin.builder()

.serviceHttpUrl("http://localhost:8080")

.build();

//获取Topic的分区分布信息

String[]partitions=admin.topics().getPartitions("persistent://my-property/use/my-ns/my-topic");

//打印分区信息

for(Stringpartition:partitions){

System.out.println("Partition:"+partition);

}

}catch(PulsarAdminExceptione){

e.printStackTrace();

}

}

}4.3.3解释此代码示例展示了如何使用Pulsar管理API来获取Topic的分区分布信息。getPartitions()方法返回一个字符串数组，包含了Topic的所有分区。通过这个信息，可以了解Topic的负载情况和资源分配状态，必要时进行手动调整。以上示例和解释详细介绍了Pulsar中Topic的高级操作，包括分区策略、持久化与非持久化Topic的区别，以及负载均衡与资源分配的管理。通过这些高级功能，可以更有效地管理和优化Pulsar消息队列的性能。5Topic的监控与故障排查5.1监控Topic的健康状态在Pulsar中，监控Topic的健康状态是确保消息队列稳定运行的关键。Pulsar提供了多种工具和API来帮助我们监控Topic的性能和状态。5.1.1使用PulsarManagerPulsarManager是一个基于Web的管理工具，可以用来查看和管理Pulsar集群。要监控Topic的健康状态，可以通过PulsarManager的Topic页面查看以下信息：消息速率：每秒发送和接收的消息数量。消息大小：每秒发送和接收的消息的平均大小。未确认消息数：当前未被消费者确认的消息数量。订阅者状态：每个订阅者的状态，包括活跃状态和未确认消息数。5.1.2使用PulsarAdminAPIPulsarAdminAPI提供了更细粒度的监控信息，可以通过编程方式访问。以下是一个使用Python和PulsarAdminAPI监控Topic状态的例子：importrequests

#设置PulsarAdminAPI的URL

admin_url="http://localhost:8080/admin/v2"

#获取Topic的统计信息

topic_stats_url=f"{admin_url}/topics/persistent/public/default/my-topic/stats"

response=requests.get(topic_stats_url)

ifresponse.status_code==200:

stats=response.json()

print("TopicStats:")

print(f"-Messagessent:{stats['msgRateIn']}")

print(f"-Messagesreceived:{stats['msgRateOut']}")

print(f"-Unackedmessages:{stats['unackedMessages']}")

else:

print("Failedtogettopicstats.")5.2故障排查与日志分析当Topic出现故障时，日志分析是定位问题的重要手段。Pulsar的日志文件包含了详细的运行信息，可以帮助我们理解系统的行为。5.2.1查看Broker日志Broker是Pulsar的核心组件，处理Topic的读写请求。当Topic出现问题时，Broker的日志通常会包含关键信息。以下是一个使用kubectl命令查看Broker日志的例子（假设在Kubernetes环境中运行Pulsar）：kubectllogs<broker-pod-name>-npulsar-cpulsar-broker在日志中，我们应关注以下信息：错误信息：任何错误或警告信息。性能指标：如CPU和内存使用情况，这可能影响Topic的性能。操作记录：如Topic的创建、删除或重新配置操作。5.2.2使用PulsarFunctions进行日志分析PulsarFunctions可以用来实时分析日志数据。例如，可以创建一个Function来监控Broker日志，当检测到特定的错误模式时发送警报。frompulsarimportFunction

classLogAnalyzer(Function):

def__init__(self):

self.error_pattern="ERROR"

defprocess(self,input,context):

ifself.error_patternininput:

#发送警报

context.publish("alert-topic",f"Errordetected:{input}")5.3性能调优与最佳实践为了确保Topic的高性能和高可用性，需要遵循一些最佳实践进行调优。5.3.1调整消息持久化策略消息持久化策略对性能有重大影响。例如，可以通过调整bookkeeper的配置来优化写入速度和存储效率。#bookkeeper配置文件示例

ledgerCacheSizeMB=1024

ledgerCacheTimeLimitMinutes=105.3.2使用多订阅者分担负载在高负载场景下，使用多个订阅者可以分担处理消息的负载，提高系统的吞吐量。#创建多个订阅者示例

frompulsarimportConsumerType

#创建订阅者

consumer1=client.subscribe("my-topic","my-subscription",consumer_type=ConsumerType.Shared)

consumer2=client.subscribe("my-topic","my-subscription",consumer_type=ConsumerType.Shared)5.3.3监控和调整资源使用定期监控Broker和bookkeeper的资源使用情况，如CPU、内存和磁盘空间。根据监控结果调整资源分配，以避免资源瓶颈。#使用top命令监控Broker资源使用

top-b-n1|greppulsar-broker通过遵循上述监控、故障排查和性能调优的最佳实践，可以有效地管理Pulsar中的Topic，确保消息队列的稳定性和高效性。6实践案例与应用场景6.1Pulsar在实时数据流处理中的应用在实时数据流处理场景中，ApachePulsar以其独特的分布式消息队列架构，提供了高吞吐量、低延迟和持久化的消息存储能力。Pulsar的Topic管理与操作在这一场景下显得尤为重要，它确保了数据的高效传输和处理。6.1.1原理Pulsar的Topic是消息的载体，每个Topic可以有多个分区，以实现水平扩展。Topic支持两种消息分发模式：Publish-Subscribe（发布-订阅）和Point-to-Point（点对点）。在实时数据流处理中，通常采用发布-订阅模式，多个订阅者可以同时消费同一Topic的消息，实现数据的并行处理。6.1.2内容创建Topic：在Pulsar中，可以通过管理工具或API创建Topic。例如，使用Pulsar的admin工具创建一个名为my-topic的Topic：bin/pulsar-admintopicscreatepersistent://my-tenant/my-namespace/my-topicTopic分区：为了提高处理能力，可以将Topic分区。例如，创建一个具有3个分区的Topic：bin/pulsar-admintopicscreate-partitioned-topicpersistent://my-tenant/my-namespace/my-topic--partitions3消息发布：生产者可以向Topic发布消息。以下是一个使用JavaAPI发布消息的示例：importorg.apache.pulsar.client.api.Message;

importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

importorg.apache.pulsar.client.api.Producer;

importorg.apache.pulsar.client.api.Schema;

publicclassMyProducer{

publicstaticvoidmain(String[]args)throwsPulsarClientException{

PulsarClientclient=PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();

Producer<String>producer=client.newProducer(Schema.STRING).topic("my-topic").create();

for(inti=0;i<10;i++){

producer.send("Message"+i);

}

producer.close();

client.close();

}

}消息消费：消费者可以订阅Topic并消费消息。以下是一个使用JavaAPI订阅并消费消息的示例：importorg.apache.pulsar.client.api.Consumer;

importorg.apache.pulsar.client.api.Message;

importorg.apache.pulsar.client.api.PulsarClient;

importorg.apache.pulsar.client.api.PulsarClientException;

importorg.apache.pulsar.client.api.Schema;

publicclassMyConsumer{

publicstatic