新手求助：基于C#实现分布式系统中Apache Pulsar跨机生产消费的指导

完全可以用C#实现你需要的跨机器消息收发架构，Pulsar官方提供了.NET客户端库，下面是具体实施步骤：

1. 前置准备 #

• 确保两台机器均可访问你的Pulsar集群（单节点Pulsar服务也支持），注意Pulsar默认端口为6650，需保证网络连通性，防火墙开放对应端口。
• 在两台机器的C#项目中安装Pulsar客户端包：
  使用NuGet包管理器：Install-Package Apache.Pulsar.Client
  或.NET CLI命令：dotnet add package Apache.Pulsar.Client

2. 生产者（Producer）代码（部署在机器A） #

using Apache.Pulsar.Client;
using Apache.Pulsar.Client.Api;

// 配置Pulsar客户端连接信息
var clientConfig = new PulsarClientConfigBuilder()
    .ServiceUrl("pulsar://[你的Pulsar服务IP]:6650") // 替换为实际Pulsar服务的局域网/公网IP
    .Build();

await using var client = PulsarClient.Builder().Config(clientConfig).Build();

// 创建指向目标主题的生产者
await using var producer = await client.NewProducer()
    .Topic("persistent://public/default/test-topic") // 主题路径，可按需修改
    .CreateAsync();

// 循环发送测试消息
for (int i = 0; i < 10; i++)
{
    var msgContent = $"跨机器测试消息 {i}";
    await producer.SendAsync(msgContent);
    Console.WriteLine($"已发送: {msgContent}");
    await Task.Delay(1000);
}

Console.WriteLine("生产者任务完成");

3. 消费者（Consumer）代码（部署在机器B） #

using Apache.Pulsar.Client;
using Apache.Pulsar.Client.Api;

// 配置与生产者相同的Pulsar连接信息
var clientConfig = new PulsarClientConfigBuilder()
    .ServiceUrl("pulsar://[你的Pulsar服务IP]:6650")
    .Build();

await using var client = PulsarClient.Builder().Config(clientConfig).Build();

// 创建订阅同一主题的消费者
await using var consumer = await client.NewConsumer()
    .Topic("persistent://public/default/test-topic") // 必须与生产者主题一致
    .SubscriptionName("test-cross-machine-sub") // 订阅名称需唯一
    .SubscriptionType(SubscriptionType.Exclusive) // 单消费者用独占模式，多消费者可改用Shared
    .MessageReceivedHandler((consumer, msg) =>
    {
        var content = System.Text.Encoding.UTF8.GetString(msg.Data);
        Console.WriteLine($"收到消息: {content}");
        // 确认消息已处理，避免重复消费
        _ = consumer.AcknowledgeAsync(msg);
    })
    .SubscribeAsync();

Console.WriteLine("消费者已启动，等待消息...");
Console.ReadLine(); // 保持程序运行

4. 关键注意事项 #

• Pulsar服务端配置：如果是本地搭建的Pulsar，需修改conf/broker.conf中的advertisedAddress为Pulsar机器的实际IP，不要用localhost，否则其他机器无法连接。
• 权限控制：新手可先关闭权限校验，修改conf/broker.conf中的authorizationEnabled=false，避免因权限问题导致收发失败。
• 主题类型：示例中使用的persistent主题会持久化消息，即使消费者离线，重新上线后仍可获取未消费的消息；若不需要持久化，可改用non-persistent前缀的主题。

内容的提问来源于stack exchange，提问作者Sushmita Velan