使用CSharp在Kafka中进行自定义序列化

using Confluent.Kafka;
using System;
using System.Text;

public class CustomSerializer<T> : ISerializer<T>, IDeserializer<T>
{
    public byte[] Serialize(T data, SerializationContext context)
    {
        // 自定义序列化逻辑
        // 将对象转换为字节数组
        string json = JsonConvert.SerializeObject(data);
        return Encoding.UTF8.GetBytes(json);
    }

    public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
    {
        // 自定义反序列化逻辑
        // 将字节数组转换为对象
        string json = Encoding.UTF8.GetString(data.ToArray());
        return JsonConvert.DeserializeObject<T>(json);
    }
}

using Confluent.Kafka;
using System;

public class KafkaExample
{
    private static readonly string KafkaBootstrapServers = "kafka-bootstrap-server:9092";
    private static readonly string KafkaTopic = "example-topic";

    public static void Main(string[] args)
    {
        var config = new ProducerConfig
        {
            BootstrapServers = KafkaBootstrapServers
        };

        // 创建生产者
        using (var producer = new ProducerBuilder<string, YourCustomObject>(config)
            .SetValueSerializer(new CustomSerializer<YourCustomObject>())
            .Build())
        {
            // 发送消息
            var message = new Message<string, YourCustomObject>
            {
                Key = "key",
                Value = new YourCustomObject()
                {
                    // 设置对象属性
                }
            };

            producer.Produce(KafkaTopic, message);
        }

        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = KafkaBootstrapServers,
            GroupId = "example-group",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };

        // 创建消费者
        using (var consumer = new ConsumerBuilder<string, YourCustomObject>(consumerConfig)
            .SetValueDeserializer(new CustomSerializer<YourCustomObject>())
            .Build())
        {
            // 订阅主题
            consumer.Subscribe(KafkaTopic);

            // 消费消息
            while (true)
            {
                var consumeResult = consumer.Consume();
                if (consumeResult != null)
                {
                    var message = consumeResult.Message;
                    Console.WriteLine($"Received message: Key={message.Key}, Value={message.Value}");
                }
            }
        }
    }
}