|
|
|
|
@ -1,5 +1,7 @@
|
|
|
|
|
using Confluent.Kafka;
|
|
|
|
|
using Microsoft.EntityFrameworkCore;
|
|
|
|
|
using MySqlConnector;
|
|
|
|
|
using Newtonsoft.Json;
|
|
|
|
|
using System;
|
|
|
|
|
using System.Threading;
|
|
|
|
|
using System.Threading.Tasks;
|
|
|
|
|
@ -8,8 +10,9 @@ namespace KafkaEFTest
|
|
|
|
|
{
|
|
|
|
|
internal class Program
|
|
|
|
|
{
|
|
|
|
|
static TimeSpan DefaultTimeout = TimeSpan.FromSeconds(30);
|
|
|
|
|
private static void Main(string[] args)
|
|
|
|
|
private static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(30);
|
|
|
|
|
|
|
|
|
|
private static void Main()
|
|
|
|
|
{
|
|
|
|
|
using (var db = new TestDbContext())
|
|
|
|
|
{
|
|
|
|
|
@ -17,31 +20,37 @@ namespace KafkaEFTest
|
|
|
|
|
}
|
|
|
|
|
Task.Run(async () =>
|
|
|
|
|
{
|
|
|
|
|
var topic = "test-topic";
|
|
|
|
|
var config = new ProducerConfig
|
|
|
|
|
{
|
|
|
|
|
BootstrapServers = "localhost:9092",
|
|
|
|
|
TransactionalId = "test",
|
|
|
|
|
};
|
|
|
|
|
while (true)
|
|
|
|
|
{
|
|
|
|
|
var entity = new TestEntity();
|
|
|
|
|
using (var db = new TestDbContext())
|
|
|
|
|
using (var dbContext = new TestDbContext())
|
|
|
|
|
{
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
using (var p = new ProducerBuilder<Null, string>(new ProducerConfig { BootstrapServers = "localhost:9092", TransactionalId = entity.Id.ToString() }).Build())
|
|
|
|
|
using (var producer = new ProducerBuilder<string, string>(config).Build())
|
|
|
|
|
{
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
p.InitTransactions(DefaultTimeout);
|
|
|
|
|
p.BeginTransaction();
|
|
|
|
|
db.Database.BeginTransaction();
|
|
|
|
|
db.Tests.Add(entity);
|
|
|
|
|
db.SaveChanges();
|
|
|
|
|
var dr = await p.ProduceAsync("test-topic", new Message<Null, string> { Value = entity.Id + ":" + entity.Value });
|
|
|
|
|
db.Database.CommitTransaction();
|
|
|
|
|
p.CommitTransaction(DefaultTimeout);
|
|
|
|
|
Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
|
|
|
|
|
var entity = new TestEntity();
|
|
|
|
|
producer.InitTransactions(DefaultTimeout);
|
|
|
|
|
producer.BeginTransaction();
|
|
|
|
|
dbContext.Database.BeginTransaction();
|
|
|
|
|
dbContext.Tests.Add(entity);
|
|
|
|
|
dbContext.SaveChanges();
|
|
|
|
|
var dr = await producer.ProduceAsync(topic, new Message<string, string> { Key = "test_table", Value = JsonConvert.SerializeObject(entity) });
|
|
|
|
|
dbContext.Database.CommitTransaction();
|
|
|
|
|
producer.CommitTransaction(DefaultTimeout);
|
|
|
|
|
Console.WriteLine($"send message offset: '{dr.TopicPartitionOffset}' value: '{dr.Value}");
|
|
|
|
|
}
|
|
|
|
|
catch (Exception ex)//DbUpdateException//ProduceException<Null,string>
|
|
|
|
|
{
|
|
|
|
|
db.Database.RollbackTransaction();
|
|
|
|
|
p.AbortTransaction(DefaultTimeout);
|
|
|
|
|
dbContext.Database.RollbackTransaction();
|
|
|
|
|
producer.AbortTransaction(DefaultTimeout);
|
|
|
|
|
Console.WriteLine(ex.Message);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@ -57,13 +66,17 @@ namespace KafkaEFTest
|
|
|
|
|
});
|
|
|
|
|
var conf = new ConsumerConfig
|
|
|
|
|
{
|
|
|
|
|
GroupId = "test-consumer-group",
|
|
|
|
|
GroupId = "group_test",
|
|
|
|
|
BootstrapServers = "localhost:9092",
|
|
|
|
|
AutoOffsetReset = AutoOffsetReset.Earliest
|
|
|
|
|
EnableAutoCommit = false,
|
|
|
|
|
StatisticsIntervalMs = 5000,
|
|
|
|
|
SessionTimeoutMs = 6000,
|
|
|
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
|
|
|
EnablePartitionEof = true
|
|
|
|
|
};
|
|
|
|
|
using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
|
|
|
|
|
using (var consumer = new ConsumerBuilder<string, string>(conf).Build())
|
|
|
|
|
{
|
|
|
|
|
c.Subscribe("test-topic");
|
|
|
|
|
consumer.Subscribe("test-topic");
|
|
|
|
|
|
|
|
|
|
CancellationTokenSource cts = new CancellationTokenSource();
|
|
|
|
|
Console.CancelKeyPress += (_, e) =>
|
|
|
|
|
@ -78,8 +91,26 @@ namespace KafkaEFTest
|
|
|
|
|
{
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
var cr = c.Consume(cts.Token);
|
|
|
|
|
Console.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
|
|
|
|
|
var consumeResult = consumer.Consume(cts.Token);
|
|
|
|
|
if (consumeResult.IsPartitionEOF)
|
|
|
|
|
{
|
|
|
|
|
Console.WriteLine(
|
|
|
|
|
$"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Console.WriteLine($"group received message offset: '{consumeResult.TopicPartitionOffset}' key: {consumeResult.Message.Key} value: '{consumeResult.Message.Value}'");
|
|
|
|
|
|
|
|
|
|
Doris(consumeResult.Value);
|
|
|
|
|
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
consumer.Commit(consumeResult);
|
|
|
|
|
}
|
|
|
|
|
catch (KafkaException e)
|
|
|
|
|
{
|
|
|
|
|
Console.WriteLine($"Commit error: {e.Error.Reason}");
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
catch (ConsumeException e)
|
|
|
|
|
{
|
|
|
|
|
@ -90,10 +121,51 @@ namespace KafkaEFTest
|
|
|
|
|
catch (OperationCanceledException)
|
|
|
|
|
{
|
|
|
|
|
// Ensure the consumer leaves the group cleanly and final offsets are committed.
|
|
|
|
|
c.Close();
|
|
|
|
|
consumer.Close();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
//create database test;
|
|
|
|
|
//CREATE USER 'usr' IDENTIFIED BY 'pwd';
|
|
|
|
|
//GRANT ALL ON test TO usr;
|
|
|
|
|
//CREATE TABLE IF NOT EXISTS test
|
|
|
|
|
//(
|
|
|
|
|
// `id` VARCHAR(64),
|
|
|
|
|
// `value` VARCHAR(128),
|
|
|
|
|
// `number` INT
|
|
|
|
|
//)
|
|
|
|
|
//UNIQUE KEY(`id`)
|
|
|
|
|
//distributed by hash(id) buckets 1
|
|
|
|
|
//properties(
|
|
|
|
|
// "replication_num" = "1"
|
|
|
|
|
//);
|
|
|
|
|
|
|
|
|
|
private static void Doris(string value)
|
|
|
|
|
{
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
var entity = JsonConvert.DeserializeObject<TestEntity>(value);
|
|
|
|
|
var connectionString = "Server=localhost;Port=9030;Database=test;Uid=usr;Pwd=pwd;";
|
|
|
|
|
using (var conn = new MySqlConnection(connectionString))
|
|
|
|
|
{
|
|
|
|
|
conn.Open();
|
|
|
|
|
using (var cmd = conn.CreateCommand())
|
|
|
|
|
{
|
|
|
|
|
cmd.CommandTimeout = 60;
|
|
|
|
|
cmd.CommandText = $"insert into test (`id`,`value`,`number`) values ('{entity.Id}','{entity.Value}',{entity.Number})";
|
|
|
|
|
cmd.ExecuteNonQuery();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
catch (Exception ex)
|
|
|
|
|
{
|
|
|
|
|
Console.WriteLine(ex.GetType().FullName);
|
|
|
|
|
//"Unknown database 'default_cluster:test'"
|
|
|
|
|
//errCode = 2, detailMessage = Unknown table 'test'
|
|
|
|
|
throw;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public class TestDbContext : DbContext
|
|
|
|
|
@ -112,9 +184,11 @@ namespace KafkaEFTest
|
|
|
|
|
{
|
|
|
|
|
this.Id = Guid.NewGuid();
|
|
|
|
|
this.Value = DateTime.Now.Ticks.ToString();
|
|
|
|
|
this.Number = new Random((int)DateTime.Now.Ticks).Next();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public Guid Id { get; set; }
|
|
|
|
|
public string Value { get; set; }
|
|
|
|
|
public int Number { get; set; }
|
|
|
|
|
}
|
|
|
|
|
}
|
