是否可以在 kafka 中发布 xml?

Question

是否可以在kafka中发布xml？如果是，可以使用哪种方法来序列化要在 kafka 中发布的数据？

Answer 1

Kafka 接受任何可序列化的输入格式。Xml 只是文本，因此可以使用纯字符串序列化程序。

否则，如果您希望在推送消息之前进行其他验证(例如检查内容实际上是 xml(，则需要编写自己的序列化程序/反序列化程序实现

至于 apache-kafka-connect，你可以参考这个 Connect 变压器 https://github.com/jcustenborder/kafka-connect-transform-xml

Answer 2

您应该对要发布和订阅的消息使用序列化程序/反序列化程序。关于 kafka 主题的内容看起来像一个字符串 xml。 下面是序列化和反序列化字符串 XML 的 C# 代码示例

C# 代码示例，用于生成和使用 Kafka 消息 confluent.io 它支持方便的编程语言

制作人

using System;
using Confluent.Kafka;
class Program
{
public static void Main(string[] args)
{
var conf = new ProducerConfig { BootstrapServers = "localhost:9092" };
Action<DeliveryReport<Null, string>> handler = r => 
Console.WriteLine(!r.Error.IsError
? $"Delivered message to {r.TopicPartitionOffset}"
: $"Delivery Error: {r.Error.Reason}");
using (var p = new ProducerBuilder<Null, string>(conf).Build())
{
// call XmlSerializeToString(this object objectInstance) from ref#1
var xmlMessage = "xml object".XmlSerializeToString();
p.Produce("my-topic", new Message<Null, string> { Value = xmlMessage}, handler);       
// wait for up to 10 seconds for any inflight messages to be delivered.
p.Flush(TimeSpan.FromSeconds(10));
}
}

}

消费者

using System;
using System.Threading;
using Confluent.Kafka;
class Program
{
public static void Main(string[] args)
{
var conf = new ConsumerConfig
{ 
GroupId = "test-consumer-group",
BootstrapServers = "localhost:9092",
// Note: The AutoOffsetReset property determines the start offset in the event
// there are not yet any committed offsets for the consumer group for the
// topic/partitions of interest. By default, offsets are committed
// automatically, so in this example, consumption will only start from the
// earliest message in the topic 'my-topic' the first time you run the program.
AutoOffsetReset = AutoOffsetReset.Earliest
};
using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
{
c.Subscribe("my-topic");
CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => {
e.Cancel = true; // prevent the process from terminating.
cts.Cancel();
};
try
{
while (true)
{
try
{
var cr = c.Consume(cts.Token);
var xmlContent = cr.Value.XmlDeserializeFromString()
Console.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
}
catch (ConsumeException e)
{
Console.WriteLine($"Error occured: {e.Error.Reason}");
}
}
}
catch (OperationCanceledException)
{
// Ensure the consumer leaves the group cleanly and final offsets are committed.
c.Close();
}
}
}
}