0

我正在尝试使用 IBM 消息中心创建生产者和消费者应用程序。对于生产者，我使用以下代码：

var config = new Dictionary<string, object> {
                { "bootstrap.servers", brokerList },
                { "group.id", "simple-csharp-producer" },
                { "client.id", "some string for id such as FR45fHth..." },
                {"api.version.request","true" },
                {"sasl.mechanisms","PLAIN" },
                {"sasl.username","the first 16 charachters of the client.id" },
                {"sasl.password","the other characters left" }
            };

            using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8)))
            {
              ....
            }

对于消费者，我正在使用类似的配置属性。

消费者的其余代码：

using (var consumer = new Consumer<Null, string>(config, null, new StringDeserializer(Encoding.UTF8)))
            {
                consumer.Assign(new List<TopicPartitionOffset> { new TopicPartitionOffset(topics, 0, 0) });

                while (true)
                {
                    Message<Null, string> msg;
                    if (consumer.Consume(out msg, TimeSpan.FromSeconds(1)))
                    {
                        Console.WriteLine($"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {msg.Value}");
                    }
                }
            }

对于生产者：

using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8)))
            {
                Console.WriteLine($"{producer.Name} producing on {topicName}. q to exit.");

                string text;
                while ((text = Console.ReadLine()) != "q")
                {
                    var deliveryReport = producer.ProduceAsync(topicName, null, text);
                    deliveryReport.ContinueWith(task =>
                    {
                        Console.WriteLine($"Partition: {task.Result.Partition}, Offset: {task.Result.Offset}");
                    });
                }

                // Tasks are not waited on synchronously (ContinueWith is not synchronous),
                // so it's possible they may still in progress here.
                producer.Flush(Convert.ToInt32(TimeSpan.FromSeconds(10)));

无论如何，它没有用，没有任何迹象表明发送任何东西......缺少什么？或者我可以用它来工作吗？

我得到的日志：

*sasl_ssl://kafka03-prod02.messagehub.services.eu-gb.bluemix.net:9093/bootstrap：无法初始化 SASL 身份验证：平台不支持 SASL 机制“PLAIN”

* 1/1 经纪人倒闭

4

2 回答 2

0

Mickael 发布的设置是正确的。

但是，如果您在 Windows 上运行 - 要获得 SASL/SSL 支持（Message Hub 需要），您需要 librdkafka 0.11

使用 librdkafka 0.9.5，您无法从 Windows 连接到 MH

于 2017-07-04T15:03:26.893 回答
0

我自己没有使用过 Confluent C# 客户端，但据我所知，它基于 librdkakfa，因此您至少需要更多配置才能连接到 Message Hub：

  • security.protocol调成SASL_SSL
  • ssl.ca.location设置为您的 CA 证书的路径
于 2017-07-04T13:22:11.290 回答