如何使用.NET Core实现CSV数据批量导入Logstash

前置准备：Logstash 输入端配置 #

首先需要和运维同学确认Logstash已经配置了对应协议的输入插件，两种最常用的配置示例如下：

HTTP输入配置（易调试，推荐） #

input {
  http {
    port => 8081 # 自定义接收端口
    codec => json # 直接解析JSON格式数据
    # 若需要鉴权可新增user/password配置
  }
}

TCP输入配置（性能更高，适合超大数据量） #

input {
  tcp {
    port => 5045 # 自定义接收端口
    codec => json_lines # 按行解析JSON格式数据
  }
}

.NET Core 实现方案 #

方案1：通过HTTP接口批量发送（推荐） #

无需引入第三方类库，直接用.NET自带的HttpClient即可实现，支持批量打包发送，调试成本极低。

1. 首先在Program.cs中注册单例HttpClient

builder.Services.AddHttpClient<ILogstashSender, LogstashSender>(client =>
{
    client.BaseAddress = new Uri("http://你的Logstash服务地址:8081/");
    client.Timeout = TimeSpan.FromSeconds(30);
    // 若Logstash开启了基础鉴权，新增以下配置
    // var authenticationBytes = Encoding.ASCII.GetBytes("用户名:密码");
    // client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", Convert.ToBase64String(authenticationBytes));
});

2. 实现批量发送逻辑

public interface ILogstashSender
{
    Task BatchSendAsync<T>(IEnumerable<T> processedCsvData);
}

public class LogstashSender : ILogstashSender
{
    private readonly HttpClient _httpClient;

    public LogstashSender(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }

    public async Task BatchSendAsync<T>(IEnumerable<T> processedCsvData)
    {
        // 直接将过滤后的CSV实体列表序列化为JSON数组批量发送
        var requestContent = JsonContent.Create(processedCsvData);
        var response = await _httpClient.PostAsync(string.Empty, requestContent);
        // 校验响应状态，异常时可结合重试逻辑处理
        response.EnsureSuccessStatusCode();
    }
}

批量大小建议控制在单请求1000~5000条，可根据单条数据大小调整，避免请求体积过大触发超时。

方案2：通过TCP协议发送（高性能场景） #

适合单次导入百万级以上数据的场景，传输效率更高。

public async Task TcpBatchSendAsync<T>(IEnumerable<T> processedCsvData, string logstashHost, int logstashPort=5045)
{
    // 超大量数据传输时可复用TcpClient实例，无需每次新建
    using var tcpClient = new TcpClient();
    await tcpClient.ConnectAsync(logstashHost, logstashPort);
    await using var stream = tcpClient.GetStream();
    await using var writer = new StreamWriter(stream) { NewLine = "\n", AutoFlush = true };
    foreach (var data in processedCsvData)
    {
        var jsonStr = System.Text.Json.JsonSerializer.Serialize(data);
        await writer.WriteLineAsync(jsonStr);
    }
}

注意事项 #

• 正式批量发送前先做单条数据连通性测试，可临时将Logstash输出配置为stdout验证数据是否正常解析
• 建议新增重试、降级逻辑，可通过Polly实现网络波动时的自动重试，避免数据丢失
• 若传输敏感数据，可协商Logstash端开启HTTPS/TLS加密传输
• 若公司Logstash统一使用Beats协议接收数据，可引入官方Beats客户端类库实现发送，逻辑和上述方案类似

内容的提问来源于stack exchange，提问作者SomeDude