Compare commits
No commits in common. "33606ca9eb26ff21f8a5cd58425ce782867479bc" and "1936795778ee352549eabda07fa0534150af8b7c" have entirely different histories.
33606ca9eb
...
1936795778
@ -12,7 +12,6 @@
|
||||
<PackageReference Include="Dapper" Version="2.1.72" />
|
||||
<PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.5" />
|
||||
<PackageReference Include="Npgsql" Version="10.0.2" />
|
||||
<PackageReference Include="Npgsql.DependencyInjection" Version="10.0.2" />
|
||||
<PackageReference Include="Swashbuckle.AspNetCore" Version="6.4.0" />
|
||||
</ItemGroup>
|
||||
|
||||
|
||||
@ -1,4 +1,3 @@
|
||||
using System.Data;
|
||||
using System.Text.Json;
|
||||
using Confluent.Kafka;
|
||||
using Npgsql;
|
||||
@ -8,17 +7,18 @@ namespace GeoPulse_Pipeline;
|
||||
public class KafkaConsumer : BackgroundService
|
||||
{
|
||||
private readonly ILogger<KafkaConsumer> _logger;
|
||||
private readonly NpgsqlDataSource _dataSource;
|
||||
private readonly string _connectionString;
|
||||
private readonly string _kafkaHost;
|
||||
|
||||
private const int BatchSize = 1000;
|
||||
private readonly TimeSpan MaxWaitTime = TimeSpan.FromSeconds(2);
|
||||
|
||||
public KafkaConsumer(ILogger<KafkaConsumer> logger, IConfiguration configuration, NpgsqlDataSource dataSource)
|
||||
public KafkaConsumer(ILogger<KafkaConsumer> logger, IConfiguration configuration)
|
||||
{
|
||||
_logger = logger;
|
||||
_connectionString = configuration.GetConnectionString("DefaultConnection")
|
||||
?? throw new ArgumentNullException("資料庫連線字串未設定");
|
||||
_kafkaHost = configuration["KafkaHost"];
|
||||
_dataSource = dataSource;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
@ -28,25 +28,20 @@ public class KafkaConsumer : BackgroundService
|
||||
var consumerConfig = new ConsumerConfig
|
||||
{
|
||||
BootstrapServers = _kafkaHost,
|
||||
GroupId = "telemetry-db-writer-group", // 💡 建議換個名字，強迫 Kafka 重置 Offset
|
||||
GroupId = "telemetry-db-writer-group",
|
||||
AutoOffsetReset = AutoOffsetReset.Earliest,
|
||||
EnableAutoCommit = false
|
||||
};
|
||||
|
||||
using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig)
|
||||
.SetErrorHandler((_, e) => _logger.LogError($"Kafka 底層錯誤: {e.Reason}"))
|
||||
.Build();
|
||||
|
||||
using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build();
|
||||
consumer.Subscribe("telemetry-events");
|
||||
|
||||
var buffer = new List<ConsumeResult<Ignore, string>>();
|
||||
var lastBatchTime = DateTime.UtcNow;
|
||||
|
||||
_logger.LogInformation("🚀 Consumer 已啟動，開始監控 Kafka 訊息...");
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
try
|
||||
{
|
||||
try // 💡 關鍵修正：try 放在 while 裡面
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
var consumeResult = consumer.Consume(TimeSpan.FromMilliseconds(500));
|
||||
|
||||
@ -60,32 +55,27 @@ public class KafkaConsumer : BackgroundService
|
||||
|
||||
if (isBatchFull || isTimeUp)
|
||||
{
|
||||
_logger.LogInformation($"[處理中] 準備批次寫入 {buffer.Count} 筆資料...");
|
||||
|
||||
await BulkInsertToDatabaseAsync(buffer, stoppingToken);
|
||||
|
||||
var offsets = buffer.GroupBy(msg => msg.TopicPartition)
|
||||
.Select(g => new TopicPartitionOffset(g.Key, g.Last().Offset + 1)).ToList();
|
||||
consumer.Commit(offsets);
|
||||
var latestOffset = buffer.Last();
|
||||
consumer.Commit(latestOffset);
|
||||
|
||||
_logger.LogInformation($"✅ [成功] 已寫入 {buffer.Count} 筆資料至 PostgreSQL。");
|
||||
_logger.LogInformation($"成功批次寫入 {buffer.Count} 筆資料至 PostgreSQL。");
|
||||
|
||||
buffer.Clear();
|
||||
lastBatchTime = DateTime.UtcNow;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// 💡 就算這裡噴錯，程式也只是印出 Log，然後回到 while 頂部繼續跑下一輪
|
||||
_logger.LogError(ex, "❌ [失敗] 批次處理發生錯誤！原因: {Message}", ex.Message);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError($"發生錯誤: {ex.Message} \n {ex.StackTrace}");
|
||||
|
||||
// 發生錯誤後，為了避免壞資料卡死 buffer 導致無窮重試，這裡建議清空
|
||||
buffer.Clear();
|
||||
lastBatchTime = DateTime.UtcNow;
|
||||
|
||||
// 稍微停 2 秒，避免在連續錯誤時瘋狂刷屏
|
||||
await Task.Delay(2000, stoppingToken);
|
||||
}
|
||||
buffer.Clear();
|
||||
}
|
||||
finally
|
||||
{
|
||||
consumer.Close();
|
||||
}
|
||||
}
|
||||
|
||||
@ -93,7 +83,8 @@ public class KafkaConsumer : BackgroundService
|
||||
CancellationToken stoppingToken)
|
||||
{
|
||||
var jsonOptions = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
|
||||
await using var batch = _dataSource.CreateBatch();
|
||||
await using var dataSource = NpgsqlDataSource.Create(_connectionString);
|
||||
await using var batch = dataSource.CreateBatch();
|
||||
|
||||
foreach (var msg in buffer)
|
||||
{
|
||||
|
||||
@ -1,7 +1,5 @@
|
||||
using System.Data;
|
||||
using System.Text.Json;
|
||||
using Confluent.Kafka;
|
||||
using Dapper;
|
||||
using GeoPulse_Pipeline;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Npgsql;
|
||||
@ -14,14 +12,12 @@ var producerConfig = new ProducerConfig()
|
||||
Acks = Acks.Leader
|
||||
};
|
||||
|
||||
var connString = builder.Configuration.GetConnectionString("DefaultConnection");
|
||||
|
||||
var kafkaProducer = new ProducerBuilder<Null, string>(producerConfig).Build();
|
||||
builder.Services.AddSingleton(kafkaProducer);
|
||||
builder.Services.AddEndpointsApiExplorer();
|
||||
builder.Services.AddSwaggerGen();
|
||||
builder.Services.AddHostedService<KafkaConsumer>();
|
||||
builder.Services.AddNpgsqlDataSource(connString);
|
||||
builder.Services.AddScoped<IDbConnection>(sp => sp.GetRequiredService<NpgsqlDataSource>().CreateConnection());
|
||||
|
||||
var app = builder.Build();
|
||||
|
||||
@ -34,44 +30,54 @@ if (app.Environment.IsDevelopment())
|
||||
app.UseHttpsRedirection();
|
||||
|
||||
app.MapPost("/api/telemetry", async (
|
||||
[FromBody] TelemetryRequest request,
|
||||
[FromQuery] bool useKafka,
|
||||
IProducer<Null, string> producer,
|
||||
IDbConnection dbConnection) =>
|
||||
[FromBody] TelemetryRequest request,
|
||||
[FromQuery] bool useKafka,
|
||||
IProducer<Null, string> producer,
|
||||
IConfiguration config) =>
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(request.DeviceId))
|
||||
return Results.BadRequest("Device ID is required.");
|
||||
if (string.IsNullOrWhiteSpace(request.DeviceId))
|
||||
return Results.BadRequest("Device ID is required.");
|
||||
|
||||
if (useKafka)
|
||||
{
|
||||
var message = JsonSerializer.Serialize(request);
|
||||
try
|
||||
{
|
||||
await producer.ProduceAsync("telemetry-events", new Message<Null, string> { Value = message });
|
||||
return Results.Ok(new { Status = "Success", Route = "Kafka", Message = "已推入消息佇列" });
|
||||
}
|
||||
catch (ProduceException<Null, string> ex)
|
||||
{
|
||||
Console.WriteLine($"Kafka Delivery failed: {ex.Error.Reason}");
|
||||
return Results.Problem("Kafka 寫入失敗");
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
try
|
||||
{
|
||||
string sql =
|
||||
@"INSERT INTO telemetry_history (id, device_id, geom, timestamp) VALUES (gen_random_uuid(), @DeviceId, @Lng, @Lat, @Timestamp";
|
||||
await dbConnection.ExecuteAsync(sql, request);
|
||||
if (useKafka)
|
||||
{
|
||||
var message = JsonSerializer.Serialize(request);
|
||||
try
|
||||
{
|
||||
await producer.ProduceAsync("telemetry-events", new Message<Null, string> { Value = message });
|
||||
return Results.Ok(new { Status = "Success", Route = "Kafka", Message = "已推入消息佇列" });
|
||||
}
|
||||
catch (ProduceException<Null, string> ex)
|
||||
{
|
||||
Console.WriteLine($"Kafka Delivery failed: {ex.Error.Reason}");
|
||||
return Results.Problem("Kafka 寫入失敗");
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
try
|
||||
{
|
||||
var connString = config.GetConnectionString("DefaultConnection");
|
||||
|
||||
return Results.Ok(new { Status = "Success", Route = "Direct-DB", Message = "已直接寫入資料庫" });
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine($"DB Insert failed: {ex.Message}");
|
||||
return Results.Problem("資料庫寫入失敗");
|
||||
}
|
||||
}
|
||||
await using var dataSource = NpgsqlDataSource.Create(connString!);
|
||||
await using var cmd = dataSource.CreateCommand(@"
|
||||
INSERT INTO telemetry_table (id, device_id, geom, timestamp)
|
||||
VALUES (gen_random_uuid(), $1, ST_SetSRID(ST_MakePoint($2, $3), 4326), $4)");
|
||||
|
||||
cmd.Parameters.AddWithValue(request.DeviceId);
|
||||
cmd.Parameters.AddWithValue(request.Lng);
|
||||
cmd.Parameters.AddWithValue(request.Lat);
|
||||
cmd.Parameters.AddWithValue(request.Timestamp);
|
||||
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
|
||||
return Results.Ok(new { Status = "Success", Route = "Direct-DB", Message = "已直接寫入資料庫" });
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Console.WriteLine($"DB Insert failed: {ex.Message}");
|
||||
return Results.Problem("資料庫寫入失敗");
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
app.Lifetime.ApplicationStopping.Register(() =>
|
||||
|
||||
@ -5,8 +5,7 @@
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Information",
|
||||
"Microsoft.AspNetCore": "Warning",
|
||||
"Npgsql": "Warning"
|
||||
"Microsoft.AspNetCore": "Warning"
|
||||
}
|
||||
},
|
||||
"AllowedHosts": "*",
|
||||
|
||||
Loading…
Reference in New Issue
Block a user