|
|
|
@ -1,6 +1,6 @@
|
|
|
|
|
using Confluent.Kafka;
|
|
|
|
|
using Microsoft.EntityFrameworkCore;
|
|
|
|
|
using MySqlConnector;
|
|
|
|
|
using MySql.Data.MySqlClient;
|
|
|
|
|
using Newtonsoft.Json;
|
|
|
|
|
using System;
|
|
|
|
|
using System.Threading;
|
|
|
|
@ -37,15 +37,19 @@ namespace KafkaEFTest
|
|
|
|
|
try
|
|
|
|
|
{
|
|
|
|
|
var entity = new TestEntity();
|
|
|
|
|
var entity1 = new TestEntity();
|
|
|
|
|
producer.InitTransactions(DefaultTimeout);
|
|
|
|
|
producer.BeginTransaction();
|
|
|
|
|
dbContext.Database.BeginTransaction();
|
|
|
|
|
dbContext.Tests.Add(entity);
|
|
|
|
|
dbContext.Tests.Add(entity1);
|
|
|
|
|
dbContext.SaveChanges();
|
|
|
|
|
var dr = await producer.ProduceAsync(topic, new Message<string, string> { Key = "test_table", Value = JsonConvert.SerializeObject(entity) });
|
|
|
|
|
var dr1 = await producer.ProduceAsync(topic, new Message<string, string> { Key = "test_table", Value = JsonConvert.SerializeObject(entity1) });
|
|
|
|
|
dbContext.Database.CommitTransaction();
|
|
|
|
|
producer.CommitTransaction(DefaultTimeout);
|
|
|
|
|
Console.WriteLine($"send message offset: '{dr.TopicPartitionOffset}' value: '{dr.Value}");
|
|
|
|
|
Console.WriteLine($"send message offset: '{dr1.TopicPartitionOffset}' value: '{dr1.Value}");
|
|
|
|
|
}
|
|
|
|
|
catch (Exception ex)//DbUpdateException//ProduceException<Null,string>
|
|
|
|
|
{
|
|
|
|
@ -61,7 +65,7 @@ namespace KafkaEFTest
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
await Task.Delay(5000);
|
|
|
|
|
await Task.Delay(20);
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
var conf = new ConsumerConfig
|
|
|
|
@ -129,6 +133,7 @@ namespace KafkaEFTest
|
|
|
|
|
//create database test;
|
|
|
|
|
//CREATE USER 'usr' IDENTIFIED BY 'pwd';
|
|
|
|
|
//GRANT ALL ON test TO usr;
|
|
|
|
|
//use test;
|
|
|
|
|
//CREATE TABLE IF NOT EXISTS test
|
|
|
|
|
//(
|
|
|
|
|
// `id` VARCHAR(64),
|
|
|
|
|