腾讯云消息队列Pulsar版从入门到精通：完整对接与实战指南

apphuang2026年07月14日 16:30:114

一、腾讯云消息队列Pulsar版概述

腾讯云消息队列Pulsar版（TDMQ for Apache Pulsar）是腾讯云基于Apache Pulsar开源项目打造的企业级分布式消息队列服务。它完全兼容开源Pulsar协议，高版本完全向下兼容低版本，这意味着您可以使用开源Pulsar的SDK无缝接入腾讯云TDMQ Pulsar服务，无需修改任何业务代码。

Apache Pulsar作为下一代云原生消息系统，具备以下核心优势：采用计算与存储分离的架构，支持无限消息堆积；提供持久化与非持久化两种消息模式；支持独占、灾备、共享和Key-Shared四种灵活订阅模式；具备跨地域复制和多租户隔离能力。腾讯云TDMQ Pulsar在此基础上，提供了开箱即用的集群管理、完善的监控告警、VPC网络隔离、跨可用区高可用等企业级特性。

TDMQ Pulsar版适用于以下典型场景：分布式系统的异步解耦、削峰填谷、事件驱动架构、日志与监控数据采集、实时流数据处理、微服务间通信等。无论是初创团队的轻量级应用，还是大型企业的核心交易系统，TDMQ Pulsar都能提供稳定可靠的消息传递保障。

二、核心概念速览

在正式对接之前，需要先理解TDMQ Pulsar的几个核心概念：

集群（Cluster）：集群是TDMQ Pulsar中最顶层的资源维度，不同集群的命名空间、Topic、角色权限等完全隔离。每个集群有独立的资源配额限制，如Topic总数、消息保留时长等。通常建议将开发测试环境与生产环境使用不同的集群进行隔离。

命名空间（Namespace）：命名空间是集群内的资源分组单位，用于将同一业务或同一团队相关的Topic组织在一起。命名空间级别可以配置消息TTL（生存时间）、消息保留策略等全局参数。

Topic：Topic是消息的主题分类，生产者将消息发送到指定的Topic，消费者从Topic订阅并消费消息。TDMQ Pulsar支持持久化主题（persistent://）和非持久化主题（non-persistent://）两种类型。

订阅（Subscription）：订阅是消费者与Topic之间的关联关系，一个Topic可以有多个订阅，每个订阅代表一组消费者的消费位点。TDMQ Pulsar支持四种订阅模式：Exclusive（独占）、Failover（灾备）、Shared（共享）和Key-Shared（按Key共享）。

角色（Role）：角色用于管理客户端访问TDMQ Pulsar服务的权限，通过为角色分配命名空间级别的生产和消费权限，实现精细化的访问控制。

三、准备工作

在开始使用TDMQ Pulsar之前，需要完成以下准备工作：

首先，准备好腾讯云账号并确保账户有足够的余额用于支付集群费用。其次，确认账号具备Pulsar资源的操作权限，如果是子账户，需要主账号进行授权。然后，准备好私有网络（VPC）和子网，且VPC与Pulsar集群必须在同一个地域，不同地域的云产品内网不通。

需要先登录腾讯云控制台，点击：腾讯云控制台，还没有账号，点击：注册后再关联，已有账号点击：登录后再关联

四、控制台操作指南

4.1 创建集群

登录TDMQ Pulsar版控制台，在左侧导航栏选择"集群管理"，选择目标地域后单击"新建集群"。在创建弹窗中需要配置以下参数：

  • 集群类型：专业集群独占物理资源，数据安全保障性高，推荐生产环境使用。
  • 计费模式：专业集群仅支持包年包月预付费模式。
  • 地域：选择与业务最靠近的地域，购买后不可更换。
  • 引擎版本：专业集群支持2.9.2版本。
  • 存储形态：支持弹性存储（按量计费）和固定存储（包年包月）两种形态。
  • 跨AZ部署：在有3个或以上可用区的地域，可选择最多3个可用区进行跨可用区部署，提升集群高可用性。
  • 私有网络：将集群接入点绑定至同地域的VPC。

4.2 创建命名空间

集群创建完成后，在控制台的"命名空间"页面，选择对应地域和集群，单击"新建"创建命名空间。命名空间名称需符合命名规则，创建后可在该命名空间下统一管理Topic和配置权限。

4.3 创建角色并授权

在"角色管理"页面选择地域和集群，单击"新建角色"，填写角色名称和说明。创建完成后，进入命名空间页面，在目标命名空间的"配置权限"中，将刚创建的角色添加进来，并分配生产和消费权限。角色密钥是客户端连接集群时进行身份认证的凭证，需要在角色管理页面复制并妥善保管。

4.4 创建Topic和订阅

在"Topic管理"页面，选择目标地域、集群和命名空间，单击"新建"创建Topic。Topic名称需要填入完整路径，格式为：persistent://集群ID/命名空间/Topic名称。创建Topic后，单击操作列的"新增订阅"，为Topic创建订阅关系，订阅名称在后续消费者代码中需要使用。

五、获取连接信息

完成控制台资源配置后，需要获取以下连接信息用于客户端接入：

  • 集群接入地址（SERVICE_URL）：在集群管理页面操作栏的"接入地址"中获取，根据网络类型选择VPC内网接入点或公网接入点。
  • 角色密钥（AUTHENTICATION）：在角色管理页面复制密钥列的值。
  • Topic完整名称：在Topic管理页面复制，格式为clusterid/namespace/topic。
  • 订阅名称：在Topic详情页的消费者页签复制。

客户端可通过三种网络方式连接Pulsar集群：同一VPC内直接访问（推荐，延迟最低）、不同VPC间通过云联网通信、公网访问（需开通公网路由，注意配置安全策略）。

六、Java SDK对接实战

6.1 安装依赖

在Java项目的pom.xml中添加Pulsar客户端依赖：

<dependency>
    <groupId>org.apache.pulsar</groupId>
    <artifactId>pulsar-client</artifactId>
    <version>3.0.8</version>
</dependency>

建议使用3.0.8及以上版本，如果使用批量收发消息功能（BatchReceive），也必须使用3.0.8及以上版本。

6.2 初始化客户端

创建PulsarClient实例是接入的第一步：

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

public class Constant {
    private static final String SERVICE_URL = "http://pulsar-xxx.tdmq-pulsar.ap-sh.public.tencenttdmq.com:8080";
    private static final String AUTHENTICATION = "eyJrZXlJZC......";

    public static PulsarClient initPulsarClient() throws PulsarClientException {
        PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl(SERVICE_URL)
                .authentication(AuthenticationFactory.token(AUTHENTICATION))
                .build();
        System.out.println(">> pulsar client created.");
        return pulsarClient;
    }
}

一个PulsarClient对应一个TCP连接，原则上建议一个进程只创建一个PulsarClient实例，避免重复创建造成资源浪费。

6.3 创建生产者并发送消息

import com.tencent.cloud.tdmq.pulsar.Constant;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import java.nio.charset.StandardCharsets;

public class SimpleProducer {
    public static void main(String[] args) throws Exception {
        PulsarClient pulsarClient = Constant.initPulsarClient();
        
        Producer<byte[]> producer = pulsarClient.newProducer()
                .topic("persistent://pulsar-xxx/sdk_java/topic1")
                .create();
        System.out.println(">> pulsar producer created.");
        
        for (int i = 0; i < 10; i++) {
            String value = "my-sync-message-" + i;
            MessageId msgId = producer.newMessage()
                    .key("key" + i)
                    .value(value.getBytes(StandardCharsets.UTF_8))
                    .send();
            System.out.println("Message sent: " + msgId);
        }
        
        producer.close();
        pulsarClient.close();
    }
}

Topic名称必须使用完整路径，持久化主题格式为persistent://集群ID/命名空间/Topic名称。

6.4 创建消费者并消费消息

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;

public class SimpleConsumer {
    public static void main(String[] args) throws Exception {
        PulsarClient pulsarClient = Constant.initPulsarClient();
        
        Consumer<byte[]> consumer = pulsarClient.newConsumer()
                .topic("persistent://pulsar-xxx/sdk_java/topic1")
                .subscriptionName("topic1_sub")
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();
        System.out.println(">> pulsar consumer created.");
        
        while (true) {
            Message<byte[]> msg = consumer.receive();
            try {
                String content = new String(msg.getData(), StandardCharsets.UTF_8);
                System.out.println("Received: " + content + ", msgId: " + msg.getMessageId());
                consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        }
    }
}

七、Python SDK对接实战

7.1 安装依赖

使用pip安装Pulsar Python客户端库：

pip install 'pulsar-client==3.1.0'

7.2 创建客户端与生产者

import pulsar

SERVICE_URL = "http://pulsar-xxx.tdmq-pulsar.ap-sh.public.tencenttdmq.com:8080"
AUTHENTICATION = "eyJrZXlJZC......"

# 创建客户端
client = pulsar.Client(
    authentication=pulsar.AuthenticationToken(AUTHENTICATION),
    service_url=SERVICE_URL
)

# 创建生产者
producer = client.create_producer(
    topic='persistent://pulsar-xxx/sdk_python/topic1'
)

# 同步发送消息
for i in range(10):
    producer.send(
        f'Hello python client, message {i}'.encode('utf-8'),
        properties={'k': 'v'},
        partition_key='yourKey'
    )

# 异步发送消息
def send_callback(send_result, msg_id):
    print(f'Message published: result:{send_result} msg_id:{msg_id}')

producer.send_async(
    'Hello python client, async msg.'.encode('utf-8'),
    callback=send_callback,
    properties={'k': 'v'}
)

producer.close()
client.close()

7.3 创建消费者

# 创建消费者
consumer = client.subscribe(
    topic='persistent://pulsar-xxx/sdk_python/topic1',
    subscription_name='sub_topic1'
)

# 消费消息
while True:
    msg = consumer.receive()
    try:
        print(f"Received message '{msg.data()}' id='{msg.message_id()}'")
        consumer.acknowledge(msg)
    except Exception:
        consumer.negative_acknowledge(msg)

八、Go SDK对接实战

8.1 安装依赖

使用go get下载Pulsar Go客户端库：

go get -u "github.com/apache/pulsar-client-go/pulsar"

8.2 创建客户端与生产者

package main

import (
    "context"
    "log"
    "time"

    "github.com/apache/pulsar-client-go/pulsar"
)

func main() {
    serviceUrl := "http://pulsar-xxx.tdmq-pulsar.ap-sh.public.tencenttdmq.com:8080"
    authentication := "eyJrZXlJZC......"

    // 创建客户端
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL:               serviceUrl,
        Authentication:    pulsar.NewAuthenticationToken(authentication),
        OperationTimeout:  30 * time.Second,
        ConnectionTimeout: 10 * time.Second,
    })
    if err != nil {
        log.Fatalf("Could not instantiate Pulsar client: %v", err)
    }
    defer client.Close()

    // 创建生产者
    producer, err := client.CreateProducer(pulsar.ProducerOptions{
        Topic: "persistent://pulsar-xxx/sdk_go/topic1",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer producer.Close()

    // 发送消息
    for i := 0; i < 10; i++ {
        _, err = producer.Send(context.Background(), &pulsar.ProducerMessage{
            Payload: []byte("hello go client, message " + string(rune(i))),
            Key:     "yourKey",
            Properties: map[string]string{"key": "value"},
        })
        if err != nil {
            log.Fatal(err)
        }
    }
}

8.3 创建消费者

// 创建消费者
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:            "persistent://pulsar-xxx/sdk_go/topic1",
    SubscriptionName: "topic1_sub",
    Type:             pulsar.Shared,
    ReceiverQueueSize: 100,
})
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

// 消费消息
for {
    msg, err := consumer.Receive(context.Background())
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Received message: %s", string(msg.Payload()))
    consumer.Ack(msg)
}

九、Node.js SDK对接实战

9.1 安装依赖

使用npm安装Pulsar Node.js客户端：

npm install pulsar-client

9.2 创建客户端并收发消息

const Pulsar = require("pulsar-client");

(async () => {
    const client = new Pulsar.Client({
        serviceUrl: "http://pulsar-xxx.tdmq-pulsar.ap-sh.public.tencenttdmq.com:8080",
        authentication: new Pulsar.AuthenticationToken({
            token: "eyJrZXlJZC......"
        }),
    });

    // 创建生产者
    const producer = await client.createProducer({
        topic: "persistent://pulsar-xxx/sdk_node/topic1",
    });

    // 发送消息
    for (let i = 0; i < 10; i++) {
        await producer.send({
            data: Buffer.from(`Hello node client, message ${i}`),
            properties: { key: "value" },
        });
    }
    await producer.close();

    // 创建消费者
    const consumer = await client.subscribe({
        topic: "persistent://pulsar-xxx/sdk_node/topic1",
        subscription: "topic1_sub",
    });

    // 消费消息
    for (let i = 0; i < 10; i++) {
        const msg = await consumer.receive();
        console.log(`Received: ${msg.getData().toString()}`);
        await consumer.acknowledge(msg);
    }
    await consumer.close();
    await client.close();
})();

十、高级特性详解

10.1 延迟消息与定时消息

TDMQ Pulsar提供了专门的API来实现定时消息和延迟消息。延迟消息允许消息在指定的延迟时间后投递给消费者，适用于订单超时处理、定时任务触发等场景。使用延迟消息时，建议与普通消息使用不同的Topic来管理。在Java中发送延迟消息的示例：

producer.newMessage()
    .value("delayed message".getBytes())
    .deliverAfter(10, TimeUnit.SECONDS)
    .send();

10.2 消息重试与死信队列

当消息消费失败时，可以调用consumer.reconsumeLater()接口将消息重新投递到重试队列。客户端内部会检查消息对应的重试次数，如果达到指定的最大重试次数，消息会被投递到死信队列。死信队列中的消息表示TDMQ Pulsar已经无法自动处理，需要人工介入处理，可以通过编写专门的客户端来订阅死信Topic进行消费。重试间隔是通过延迟消息实现的。

10.3 跨可用区高可用部署

TDMQ Pulsar专业版支持跨可用区部署，在有3个或以上可用区的地域购买集群时，最多可以选择3个可用区进行部署。这种部署方式下，实例的分区副本会强制分布在各个可用区的节点上，当单个可用区不可用时，集群仍能正常提供服务。跨可用区部署是生产环境保障高可用的重要手段。

10.4 VPC网络配置与安全组

在购买集群时指定了VPC后，默认仅该VPC内的资源可以访问TDMQ Pulsar服务。如果其他VPC需要访问，可以通过配置VPC网络路由策略将目标VPC绑定至服务。Pulsar专业版还支持为VPC接入点绑定安全组，实现网络层级的IP访问控制，结合Pulsar原生的角色权限管理（ACL），可以实现网络准入与资源鉴权的双重保障。

10.5 监控与告警配置

TDMQ Pulsar基于腾讯云可观测平台提供了全面的监控告警能力。可以在集群详情页的"监控信息"页签查看多维度的监控指标，涵盖集群、命名空间、Topic等维度。告警规则配置入口有两个：一是在监控页面直接配置，二是在腾讯云可观测平台配置。当监控指标达到设定的告警阈值时，系统会通过邮件、短信、微信或电话等方式通知您，帮助及时发现问题并采取措施。

十一、客户端最佳实践

以下是基于生产实践总结的客户端使用最佳实践：

  • 客户端复用：一个PulsarClient对应一个TCP连接，原则上建议一个进程只创建一个PulsarClient实例。生产者和消费者都是可复用的资源，通过名称作为唯一标识进行区分。
  • 超时配置：建议将超时时间至少设置为5秒以上，非独占集群建议设置10秒以上，防止网络抖动影响发送成功率。
  • 连接数控制：单个Topic的生产者上限为1000个，消费者上限为2000个。如果达到上限，需要提交工单进行评估调整。
  • 消息堆积处理：如果Topic出现消息堆积超过10GB（虚拟集群限制），可以采取以下措施：排查客户端未ACK的原因并重启客户端、加大消费者数量提高消费速度、缩短命名空间的TTL时间、或升级到专业集群（无10GB限制）。
  • 版本选择：Java客户端建议使用3.0.8及以上版本，Go客户端推荐0.13.1及以上版本。2.x版本Java SDK若配置了死信队列策略，会默认创建30秒的ackTimeout自动重推配置，可能影响对重推敏感的业务。

十二、常见问题解答

问：消息轨迹中只有生产记录没有推送记录是什么原因？

可能的原因包括：曾经修改或重置了订阅的Offset导致消费位点超过该消息；命名空间TTL过短导致消息在推送前被自动确认；延迟消息的延迟时间大于TTL配置；消息的Tag与订阅不一致被过滤；Key-Shared模式下前序消息未被ACK导致后续消息无法推送。

问：生产者频繁断连重连并产生大量重复消息怎么办？

最常见的原因是生产的延迟消息延迟时间超过10天，低版本SDK对延迟超过10天的报错解析不完善导致重复生产。建议排查日志中是否出现"DELAY TIME IS TOO LONG"报错，并检查代码中延迟消息相关配置。

问：如何提升消息消费速度？

可以通过以下方式提升消费速度：增加消费者数量（注意Topic和订阅的消费者上限）；调整ReceiverQueueSize参数增加单分区预拉取消息量；检查是否存在消息堆积并排查未ACK的原因；如果是专业集群且流量特别大，考虑调大Topic分区数量。

问：TDMQ Pulsar支持哪些计费模式？

TDMQ Pulsar版支持包年包月（预付费）和按量计费（后付费）两种计费模式。专业集群支持包年包月模式，虚拟集群支持线性计价。存储费用按三副本计费，实际计费存储大小约为消息总大小的3倍。

问：客户端连接Pulsar集群有哪几种网络方式？

客户端可通过三种网络方式连接：同一VPC内直接访问（推荐，延迟最低且免费）、不同VPC间通过云联网通信、公网访问（需开通公网路由并配置安全策略）。

问：如何保障TDMQ Pulsar的数据安全？

可以从多个层面保障数据安全：网络层面使用VPC隔离和安全组IP访问控制；认证层面使用角色密钥进行身份认证；传输层面开启TLS加密，支持单向认证和双向认证；权限层面通过命名空间级别的角色权限控制生产和消费。

相关文章

腾讯云服务器购买优惠！3 个省钱攻略 + 1 个安全真相，新手必看！

腾讯云服务器购买优惠！3 个省钱攻略 + 1 个安全真相，新手必看！

最近后台总收到小伙伴私信：“腾讯云服务器看着挺好，但价格有点顶，学生党 / 小团队实在买不起咋办？” 别急！今天就来手把手教你 “花小钱办大事”，不光有省钱攻略，还会扒一扒大家最关心的安全问题，看完这…

After 10 Years as a Tencent Cloud Agent, Let Me Talk About Rebates

After 10 Years as a Tencent Cloud Agent, Let Me Talk About Rebates

Lately, I’ve been getting a lot of questions from friends: “Does Tencent offer rebates? Can you…

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

一、腾讯云代理商返利机制核心逻辑1. 行业背景与代理模式腾讯云作为国内公有云市场的第二大领导者（据IDC 2025年数据，占据国内27.6%的市场份额），采用渠道商代理模式拓展市场。代理商负…

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

2026腾讯云代理商返利政策深度解析：头部代理合作指南与成本优化策略

一、腾讯云代理商返利机制核心逻辑1. 行业背景与代理模式腾讯云作为国内公有云市场的第二大领导者（据IDC 2025年数据，占据国内27.6%的市场份额），采用渠道商代理模式拓展市场。代理商负…

2026腾讯云代理商返佣政策全解析：五级代理体系与企业上云成本优化指南

2026腾讯云代理商返佣政策全解析：五级代理体系与企业上云成本优化指南

一、腾讯云五级代理体系：权益阶梯与合作价值1. 五级代理的核心权益差异腾讯云按规模、服务能力与合作深度，构建了从基础到顶级的五级代理体系，各级权益呈现显著阶梯差：•标准级代理：入门门槛最低，仅能提供基…

2026年腾讯云代理深度解析：从折扣体系到最优合作策略

2026年腾讯云代理深度解析：从折扣体系到最优合作策略

上海汪远信息科技有限公司作为腾讯云全国级殿堂级代理，凭借13年云服务经验与深厚的官方合作关系，为企业提供全方位的上云支持，可百度：上海汪远信息科技有限公司，微信：791201210一、腾讯云代理体系全…

上海汪远信息科技有限所在公司从事多年公有云和电子签代理，能持续为用户提供商务价值和技术支持，目前从事的代理有：阿里云，腾讯云，华为云，火山云，天翼云，移动云，亚马逊云，谷歌云，微软云，腾讯电子签，法大大电子签，爱签电子签等！

网站备案号：沪ICP备15020509号-2 公安备案号：沪公网安备31011202008721号