重构异步执行方法

Posted 2020-11-05 学习园地

 public static class TeraDataHelper
    {

		//<add key="CommandTimeout" value="200000" />
		//<add key="Teradata_Connect_String" value="Data Source = 172.20.70.73;User ID = pu_dw_user_web;Password = pu_dw_user_web;"/>

        public static string TDConnstring = ConfigurationManager.ConnectionStrings["Teradata_Connect_String"].ConnectionString;
        public static int CommandTimeout = int.Parse(ConfigurationManager.AppSettings["CommandTimeout"]);

        /// <summary>
        /// TD中查询sql
        /// </summary>
        /// <param name="sqlstr">sql语句</param>
        /// <returns>返回DataRow</returns>
        public static DataSet GetRows(string sqlString)
        {
            DataSet ds = new DataSet();
            try
            {
                TdDataAdapter adapter = new TdDataAdapter(sqlString, TDConnstring);
                adapter.Fill(ds);
            }
            catch (Exception ex)
            {
                //log.writeLog("执行sql 报错,报错信息为：" + ex.Message + ", 报错语句为：‘" + sqlString + "’", "失败");
                //MessageBox.Show(ex.Message);
                //throw;
            }
            return ds;
        }

        /// <summary>

        /// 获取TD的数据

        /// </summary>

        /// <param name="sqlstr">sql语句：查询语句</param>

        /// <returns>返回数据表 DataTable</returns>
        public static DataTable GetTable(string sqlStr)
        {
            TdConnection con = new TdConnection();
            DataTable dataTable = new DataTable();
            con.ConnectionString = TDConnstring;
            try
            {
                TdCommand cmd = con.CreateCommand();
                cmd.CommandTimeout = CommandTimeout;
                cmd.CommandText = sqlStr;

                // Create the TdDataAdapter object. It retrieves the data from database and fill a single data table 
                // with in a dataset. It also capable of reconciling the changes to database.                    
                TdDataAdapter adapter = new TdDataAdapter();
                adapter.ReturnProviderSpecificTypes = true;
                adapter.SelectCommand = cmd;


                // Create a DataTable object and it represents one table of in-memory data.               

                adapter.ReturnProviderSpecificTypes = true;
                adapter.Fill(dataTable);
                return dataTable;
            }
            catch (Exception ex)
            {
                //log.writeLog("执行sql 报错,报错信息为：" + ex.Message + ", 报错语句为：‘" + sqlStr + "’", "失败");
                //MessageBox.Show(ex.Message);
                throw;
            }
            finally
            {
                if (con.State == ConnectionState.Open)
                {
                    con.Close();
                }
            }
            
        }
        public static void ExecuteDataReader(string sql, Action<TdDataReader> action)
        {
            using (var connection = new TdConnection(TDConnstring))
            {
                connection.Open();
                using (var cmd = connection.CreateCommand())
                {
                    cmd.CommandText = sql;
                    using (var dr = cmd.ExecuteReader())
                    {
                        while (dr.Read())
                                action.Invoke(dr);
                    }
                }
            }
        }
        public static async void ExecuteDataReaderAsync(string sql, Action<TdDataReader> action)
        {
            using (var connection = new TdConnection(TDConnstring))
            {
                await connection.OpenAsync();
                using (var cmd = connection.CreateCommand())
                {
                    cmd.CommandText = sql;
                    var dr = await cmd.ExecuteReaderAsync( CommandBehavior.CloseConnection);
                   
                        while (await dr.ReadAsync())
                            action.Invoke(dr);
                   
                }
            }
        }
       
        public static IEnumerable<T> ExecuteDataReader<T>(string sql,  Func<TdDataReader, T> action)
        {
            using (var connection = new TdConnection(TDConnstring))
            {
                connection.Open();
                using (var cmd = connection.CreateCommand())
                {
                    cmd.CommandText = sql;
                   using (var dr = cmd.ExecuteReader())
                    {
                        while (dr.Read())
                            yield return action.Invoke(dr);
                    }
                }
            }
        }

        public static async Task<IEnumerable<T>> ExecuteDataReaderAsync<T>(string sql, Func<TdDataReader, T> action)
        {
            using (var connection = new TdConnection(TDConnstring))
            {
                await connection.OpenAsync();
                using (var cmd = connection.CreateCommand())
                {
                    cmd.CommandText = sql;
                    using (var dr = await cmd.ExecuteReaderAsync())
                    {
                       return dr.Select(r => action(r)).ToList();
                    }
                }
            }
        }
       


    }
    public static class Extensions
    {

        public static IEnumerable<T> Select<T>(
            this TdDataReader reader, Func<TdDataReader, T> action)
        {

            while (reader.Read())
            {
                yield return action(reader);
            }
        }
    }