ADO.NET+kafka实现发布订阅保存到数据库

using System;

using System.Data;

using System.Data.SqlClient;

using Confluent.Kafka;

using Confluent.Kafka.Admin;

class Program

{

static void Main(string[] args)

{

// Kafka配置

var config = new ProducerConfig { BootstrapServers = "localhost:9092" };

var producer = new ProducerBuilderstring>(config).Build();

// 数据库配置

var sqlConnectionString = "Data Source=(local);Initial Catalog=YourDatabase;Integrated Security=True";

// Kafka主题

var topic = "your_topic";

// 发送消息到Kafka

var message = new Messagestring> { Key = Null, Value = "Hello, Kafka!" };

producer.ProduceAsync(topic, message).Wait();

Console.WriteLine("Message sent to Kafka.");

// Kafka消费者配置

var consumerConfig = new ConsumerConfig

{

BootstrapServers = "localhost:9092",

GroupId = "your_group_id",

AutoOffsetReset = AutoOffsetReset.Earliest

};

using (var consumer = new ConsumerBuilderstring>(consumerConfig).Build())

{

consumer.Subscribe(topic);

try

{

while (true)

{

try

{

var result = consumer.Consume(TimeSpan.FromSeconds(1));

string value = result.Value;

// 使用ADO.NET将消息保存到数据库

using (var sqlConnection = new SqlConnection(sqlConnectionString))

{

sqlConnection.Open();

using (var sqlCommand = new SqlCommand("INSERT INTO YourTable (MessageColumn) VALUES (@Message)", sqlConnection))

{

sqlCommand.Parameters.AddWithValue("@Message", value);

sqlCommand.ExecuteNonQuery();

}

}

Console.WriteLine($"Message '{value}' received and saved to database.");

}

catch (ConsumeException e)

{

Console.WriteLine($"Error occurred: {e.Error.Reason}");

}

}

}

catch (OperationCanceledException)

{

// 确保消费者优雅地关闭

consumer.Close();

}

}

}

}