using System;
using CncCollector.Jobs;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using Dapper;
using MySqlConnector;
using CncModels.Entity;
using CncCollector.Config;
using log4net;
namespace CncCollector.Core
{
///
/// 采集引擎主控。
/// 负责加载采集地址、管理工作线程、心跳上报、配置轮询和日终汇总调度。
///
public class CollectorEngine
{
private static readonly ILog _log = LogManager.GetLogger(typeof(CollectorEngine));
private readonly CollectorConfig _config;
private readonly ConcurrentDictionary _workers = new ConcurrentDictionary();
private readonly ProductionTracker _tracker;
// 复用的分析引擎实例(简单实现:按地址注入,避免跨线程问题)
private readonly AnalysisEngine _analysisEngine;
private readonly DailySummaryJob _dailySummary;
private Timer _heartbeatTimer;
private Timer _configPollTimer;
private Timer _dailySummaryTimer;
private Timer _logCleanupTimer;
private LogCleanupJob _logCleanupJob;
private DateTime _startTime;
private long _totalSuccess;
private long _totalFail;
private volatile bool _isRunning;
private DateTime _lastSummaryDate = DateTime.MinValue;
/// 是否运行中
public bool IsRunning => _isRunning;
/// 启动时间
public DateTime StartTime => _startTime;
/// 运行时长(秒)
public long UptimeSeconds => _isRunning ? (long)(DateTime.Now - _startTime).TotalSeconds : 0;
/// 工作线程数量
public int WorkerCount => _workers.Count;
///
/// 初始化采集引擎
///
/// 全局配置
public CollectorEngine(CollectorConfig config)
{
_config = config;
_tracker = new ProductionTracker(config.BusinessConnection);
_dailySummary = new DailySummaryJob(config.BusinessConnection);
// 初始化分析引擎(与业务库和日志库同源,后续按需调整)
_analysisEngine = new AnalysisEngine(config.BusinessConnection, config.LogConnection);
}
///
/// 启动采集引擎:加载地址、启动工作线程、启动定时任务
///
public void Start()
{
if (_isRunning) return;
_log.Info("===== 采集引擎启动 =====");
_startTime = DateTime.Now;
_isRunning = true;
// 0. 立即写入running心跳(避免启动后仪表盘读不到running状态)
WriteHeartbeat("running");
// 1. 加载并启动采集地址
LoadAndStartWorkers();
// 2. 启动心跳上报定时器
_heartbeatTimer = new Timer(OnHeartbeat, null,
TimeSpan.FromSeconds(_config.HeartbeatIntervalSeconds),
TimeSpan.FromSeconds(_config.HeartbeatIntervalSeconds));
// 3. 启动配置轮询定时器
_configPollTimer = new Timer(OnConfigPoll, null,
TimeSpan.FromSeconds(_config.ConfigPollIntervalSeconds),
TimeSpan.FromSeconds(_config.ConfigPollIntervalSeconds));
// 4. 启动日终汇总检查定时器(每分钟检查一次)
_dailySummaryTimer = new Timer(OnDailySummaryCheck, null,
TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(1));
// 5. 启动日志清理定时器(从配置读取间隔,0 表示不启用)
_logCleanupJob = new LogCleanupJob(_config.LogConnection, _config);
if (_config.LogCleanupIntervalMinutes > 0)
{
_logCleanupTimer = new Timer(OnLogCleanup, null,
TimeSpan.FromMinutes(_config.LogCleanupIntervalMinutes),
TimeSpan.FromMinutes(_config.LogCleanupIntervalMinutes));
}
_log.Info($"===== 采集引擎已启动({_workers.Count}个采集地址)=====");
}
///
/// 停止采集引擎
///
public void Stop()
{
if (!_isRunning) return;
_log.Info("===== 采集引擎停止中 =====");
_isRunning = false;
// 停止所有工作线程
foreach (var kvp in _workers)
{
kvp.Value.Stop();
}
_workers.Clear();
// 结账所有活跃段
_tracker.Dispose();
// 停止定时器
_heartbeatTimer?.Dispose();
_configPollTimer?.Dispose();
_dailySummaryTimer?.Dispose();
_logCleanupTimer?.Dispose();
// 写入停止状态心跳
WriteHeartbeat("stopped");
_log.Info("===== 采集引擎已停止 =====");
}
///
/// 刷新配置:重新加载采集地址,处理新增/删除/变更
///
public void Refresh()
{
_log.Info("刷新采集配置...");
LoadAndStartWorkers();
_log.Info("采集配置刷新完成");
}
///
/// 获取引擎状态摘要
///
public Dictionary GetStatus()
{
var status = new Dictionary
{
["isRunning"] = _isRunning,
["startTime"] = _startTime.ToString("yyyy-MM-dd HH:mm:ss"),
["uptimeSeconds"] = UptimeSeconds,
["workerCount"] = _workers.Count,
["totalSuccess"] = _totalSuccess,
["totalFail"] = _totalFail
};
var workerList = new List>();
foreach (var kvp in _workers)
{
workerList.Add(new Dictionary
{
["addressId"] = kvp.Key,
["name"] = kvp.Value.AddressName,
["url"] = kvp.Value.AddressUrl,
["isRunning"] = kvp.Value.IsRunning
});
}
status["workers"] = workerList;
return status;
}
///
/// 手动执行日终汇总(指定日期)
///
/// 要汇总的日期
public void RunDailySummary(DateTime summaryDate)
{
_log.Info($"手动触发日终汇总(日期={summaryDate:yyyy-MM-dd})");
_dailySummary.Execute(summaryDate);
}
///
/// 加载启用的采集地址并启动工作线程
///
private void LoadAndStartWorkers()
{
try
{
List addresses;
using (var conn = new MySqlConnection(_config.BusinessConnection))
{
addresses = conn.Query(
"SELECT id as Id, name as Name, url as Url, brand_id as BrandId, collect_interval as CollectInterval, is_enabled as IsEnabled, last_collect_time as LastCollectTime, last_collect_status as LastCollectStatus, fail_count as FailCount, created_at as CreatedAt, updated_at as UpdatedAt FROM cnc_collect_address WHERE is_enabled = 1"
).AsList();
}
// 停止已删除的地址
foreach (var kvp in _workers)
{
bool exists = false;
foreach (var addr in addresses)
{
if (addr.Id == kvp.Key) { exists = true; break; }
}
if (!exists)
{
kvp.Value.Stop();
CollectWorker removed;
_workers.TryRemove(kvp.Key, out removed);
_log.Info($"已停止删除的采集地址: {kvp.Value.AddressName}");
}
}
// 启动新增的地址
foreach (var addr in addresses)
{
if (!_workers.ContainsKey(addr.Id))
{
var worker = new CollectWorker(addr, _config, _tracker,
_analysisEngine, _config.BusinessConnection, _config.LogConnection);
worker.Start();
_workers[addr.Id] = worker;
_log.Info($"已启动采集地址: {addr.Name}(URL={addr.Url}, 间隔={addr.CollectInterval}秒)");
}
}
}
catch (Exception ex)
{
_log.Error("加载采集地址失败", ex);
}
}
///
/// 心跳上报回调
///
private void OnHeartbeat(object state)
{
try
{
// 统计成功/失败次数
long success = 0, fail = 0;
foreach (var kvp in _workers)
{
success += kvp.Value.SuccessCount;
fail += kvp.Value.FailCount;
}
_totalSuccess = success;
_totalFail = fail;
WriteHeartbeat("running");
}
catch (Exception ex)
{
_log.Error("心跳上报失败", ex);
}
}
///
/// 写入心跳记录到 log_collector_heartbeat
///
private void WriteHeartbeat(string status)
{
try
{
using (var conn = new MySqlConnection(_config.LogConnection))
{
conn.Execute(@"INSERT INTO log_collector_heartbeat (service_id, status, last_collect_time, success_count, fail_count, uptime_seconds, detail, created_at)
VALUES (@ServiceId, @Status, @LastCollectTime, @SuccessCount, @FailCount, @UptimeSeconds, @Detail, NOW())",
new
{
ServiceId = _config.ServiceId,
Status = status,
LastCollectTime = DateTime.Now,
SuccessCount = _totalSuccess,
FailCount = _totalFail,
UptimeSeconds = UptimeSeconds,
Detail = (string)null
});
}
}
catch (Exception ex)
{
_log.Error("写入心跳记录失败", ex);
}
}
///
/// 配置轮询回调:检查是否有配置变更
///
private void OnConfigPoll(object state)
{
try
{
// 重新从DB加载运行时配置
ConfigLoader.LoadRuntimeConfig(_config.BusinessConnection, _config);
// 重新加载采集地址
LoadAndStartWorkers();
}
catch (Exception ex)
{
_log.Error("配置轮询失败", ex);
}
}
///
/// 日终汇总检查回调:检查是否到了配置的汇总时间
///
private void OnDailySummaryCheck(object state)
{
try
{
// 解析汇总时间
TimeSpan summaryTime;
if (!TimeSpan.TryParse(_config.DailySummaryTime, out summaryTime))
{
summaryTime = new TimeSpan(1, 0, 0); // 默认01:00
}
var now = DateTime.Now;
var targetTime = new DateTime(now.Year, now.Month, now.Day, summaryTime.Hours, summaryTime.Minutes, 0);
// 检查是否到了汇总时间(±1分钟内)
if (Math.Abs((now - targetTime).TotalMinutes) <= 1)
{
// 汇总昨天的数据
DateTime summaryDate = now.Date.AddDays(-1);
if (_lastSummaryDate != summaryDate)
{
_lastSummaryDate = summaryDate;
_dailySummary.Execute(summaryDate);
}
}
}
catch (Exception ex)
{
_log.Error("日终汇总检查失败", ex);
}
}
///
/// 日志清理定时回调
///
private void OnLogCleanup(object state)
{
try
{
_logCleanupJob?.Execute();
}
catch (Exception ex)
{
_log.Error("日志清理任务执行失败", ex);
}
}
}
}