如何在Golang中实现基于RabbitMQ动态数据源的动态模型创建与取数

核心思路 #

无需预定义实体模型，通过获取结果集元数据动态解析数据，再根据目标存储类型（数据库表/Redis）完成写入。

步骤1：接收并解析RabbitMQ消息 #

用RabbitMQ客户端接收消息，解析出核心任务信息：数据库类型（SQL Server/Oracle）、源连接字符串、执行语句（或存储过程+参数）、目标存储类型及连接信息。

示例（C#）：

// 初始化RabbitMQ连接
var factory = new ConnectionFactory() { HostName = "your-rabbitmq-host" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
channel.QueueDeclare(queue: "dynamic-data-queue", durable: true, exclusive: false, autoDelete: false);

// 注册消息消费回调
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (_, ea) =>
{
    var messageBody = Encoding.UTF8.GetString(ea.Body.ToArray());
    // 反序列化消息为任务结构体
    var task = JsonSerializer.Deserialize<DataSyncTask>(messageBody);
    
    // 执行数据同步
    ExecuteDataSync(task);
    
    // 手动确认消息已处理
    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
channel.BasicConsume(queue: "dynamic-data-queue", autoAck: false, consumer: consumer);

// 自定义任务结构体
public class DataSyncTask
{
    public string DbType { get; set; } // "SqlServer"或"Oracle"
    public string SourceConnString { get; set; }
    public string ExecuteText { get; set; } // SQL语句或存储过程名
    public bool IsStoredProcedure { get; set; }
    public Dictionary<string, object> Params { get; set; }
    public string TargetType { get; set; } // "Database"或"Redis"
    public string TargetConnString { get; set; }
    public string TargetTable { get; set; } // 目标表名（数据库存储时用）
}

步骤2：动态执行查询并提取元数据 #

通过数据库连接执行语句，利用DataReader.GetSchemaTable()获取结果集的列名、数据类型等元数据，将数据转为通用字典列表存储。

示例：

private void ExecuteDataSync(DataSyncTask task)
{
    IDbConnection conn = task.DbType switch
    {
        "SqlServer" => new SqlConnection(task.SourceConnString),
        "Oracle" => new OracleConnection(task.SourceConnString),
        _ => throw new NotSupportedException("不支持的数据库类型")
    };

    try
    {
        conn.Open();
        using var cmd = conn.CreateCommand();
        cmd.CommandText = task.ExecuteText;

        // 处理存储过程参数
        if (task.IsStoredProcedure)
        {
            cmd.CommandType = CommandType.StoredProcedure;
            foreach (var param in task.Params)
            {
                var dbParam = cmd.CreateParameter();
                dbParam.ParameterName = param.Key;
                dbParam.Value = param.Value ?? DBNull.Value;
                cmd.Parameters.Add(dbParam);
            }
        }

        // 读取结果集和元数据
        using var reader = cmd.ExecuteReader();
        var schema = reader.GetSchemaTable();
        var columns = schema.AsEnumerable()
            .Select(row => new
            {
                ColumnName = row["ColumnName"].ToString(),
                DataType = Type.GetType(row["DataType"].ToString()) ?? typeof(object)
            })
            .ToList();

        // 转换数据为字典列表
        var dataRows = new List<Dictionary<string, object>>();
        while (reader.Read())
        {
            var rowDict = new Dictionary<string, object>();
            foreach (var col in columns)
            {
                rowDict[col.ColumnName] = reader[col.ColumnName] == DBNull.Value ? null : reader[col.ColumnName];
            }
            dataRows.Add(rowDict);
        }

        // 写入目标存储
        if (task.TargetType == "Database")
        {
            WriteToDatabase(task, columns, dataRows);
        }
        else if (task.TargetType == "Redis")
        {
            WriteToRedis(task, dataRows);
        }
    }
    catch (Exception ex)
    {
        // 记录异常日志，可将失败消息转入死信队列
        Console.WriteLine($"任务执行失败：{ex.Message}");
    }
    finally
    {
        conn.Close();
    }
}

步骤3：写入目标存储 #

3.1 写入数据库表

根据元数据动态创建目标表（若不存在），用批量插入提升性能。

示例：

private void WriteToDatabase(DataSyncTask task, List<dynamic> columns, List<Dictionary<string, object>> dataRows)
{
    using var targetConn = new SqlConnection(task.TargetConnString);
    targetConn.Open();

    // 动态生成建表语句（SQL Server示例）
    var createTableSql = $"IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = '{task.TargetTable}') " +
                         $"CREATE TABLE {task.TargetTable} (" +
                         $"{string.Join(", ", columns.Select(c => $"{c.ColumnName} {MapDotNetTypeToSql(c.DataType)}"))})";
    using var createCmd = targetConn.CreateCommand();
    createCmd.CommandText = createTableSql;
    createCmd.ExecuteNonQuery();

    // 批量插入数据
    using var bulkCopy = new SqlBulkCopy(targetConn);
    foreach (var col in columns)
    {
        bulkCopy.ColumnMappings.Add(col.ColumnName, col.ColumnName);
    }

    // 转换为DataTable
    var dataTable = new DataTable();
    foreach (var col in columns)
    {
        dataTable.Columns.Add(col.ColumnName, col.DataType);
    }
    foreach (var row in dataRows)
    {
        var dtRow = dataTable.NewRow();
        foreach (var col in columns)
        {
            dtRow[col.ColumnName] = row[col.ColumnName] ?? DBNull.Value;
        }
        dataTable.Rows.Add(dtRow);
    }

    bulkCopy.WriteToServer(dataTable);
}

// .NET类型到SQL Server类型的映射
private string MapDotNetTypeToSql(Type type)
{
    return type switch
    {
        Type t when t == typeof(int) => "INT",
        Type t when t == typeof(long) => "BIGINT",
        Type t when t == typeof(string) => "NVARCHAR(MAX)",
        Type t when t == typeof(DateTime) => "DATETIME2",
        Type t when t == typeof(decimal) => "DECIMAL(18,2)",
        _ => "NVARCHAR(MAX)"
    };
}

3.2 写入Redis

将数据序列化为JSON，根据业务选择Redis数据结构（List/Hash等）。

示例：

private void WriteToRedis(DataSyncTask task, List<Dictionary<string, object>> dataRows)
{
    using var redis = ConnectionMultiplexer.Connect(task.TargetConnString);
    var db = redis.GetDatabase();

    // 示例1：用List存储所有结果（每条数据为JSON字符串）
    var redisKey = $"sync:results:{Guid.NewGuid()}";
    foreach (var row in dataRows)
    {
        var json = JsonSerializer.Serialize(row);
        db.ListRightPush(redisKey, json);
    }

    // 示例2：用Hash存储单条数据（假设数据包含唯一ID列）
    // foreach (var row in dataRows)
    // {
    //     if (row.TryGetValue("Id", out var idObj))
    //     {
    //         var id = idObj.ToString();
    //         foreach (var kvp in row)
    //         {
    //             db.HashSet($"sync:data:{id}", kvp.Key, kvp.Value?.ToString() ?? string.Empty);
    //         }
    //     }
    // }
}

关键注意事项 #

• SQL注入防护：存储过程必须用参数化；若需拼接动态SQL（如表名），需做白名单校验，禁止直接拼接用户可控内容。
• 异常与重试：对数据库连接、执行失败的任务，记录日志并转入死信队列，避免消息丢失。
• 性能优化：数据库用批量插入（如SqlBulkCopy），Redis用管道（db.CreateBatch()）减少网络开销。
• 类型映射：针对Oracle等数据库，需调整类型映射逻辑（如Oracle的VARCHAR2对应.NET的string，NUMBER对应decimal）。

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