Quickie example of using BlockingCollection with multiple readers and one writer.

DatabaseCopyEngine.cs

// /*
//    DatabaseCopyEngine.cs
//    Copyright 2013 Gibraltar Software, Inc.
//    
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
// 
//        http://www.apache.org/licenses/LICENSE-2.0
// 
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
// */


using System;
using System.Collections.Concurrent;
using System.Data;
using System.Data.SqlClient;
using System.Threading.Tasks;
using Gibraltar.Analyst.Data;
using Gibraltar.Monitor;

namespace Gibraltar.Data
{
    /// <summary>
    /// Copies database items between databases
    /// </summary>
    public static class DatabaseCopyEngine
    {
        private const int MaxDataParallelism = 12; //somewhat experimentally validated to produce nearly the best throughput without overwhelming the server.
        private const string LogCategory = "Loupe.Data.Database Copy";

        /// <summary>
        /// This method will copy the data in a table 
        /// from one database to another. The
        /// source and destination can be from any type of 
        /// .NET database provider.
        /// </summary>
        /// <param name="source">Source database connection</param>
        /// <param name="destination">Destination database connection</param>
        /// <param name="tableName">The name of the table to copy</param>
        public static void CopyTable(IDbConnection source, IDbConnection destination, string tableName)
        {
            CopyTable(source, destination, "SELECT * FROM " + tableName, tableName);
        }

        /// <summary>
        /// This method will copy the data in a table 
        /// from one database to another. The
        /// source and destination can be from any type of 
        /// .NET database provider.
        /// </summary>
        /// <param name="source">Source database connection</param>
        /// <param name="destination">Destination database connection</param>
        /// <param name="sourceSQL">Source SQL statement</param>
        /// <param name="destinationTableName">Destination table name</param>
        public static void CopyTable(IDbConnection source, IDbConnection destination, string sourceSQL, string destinationTableName)
        {
            var pendingCommandQueue = new BlockingCollection<IDbCommand>(100);
            using(new OperationMetric(LogCategory, "Copying table " + destinationTableName, "Starting copying table {0} to database {1}", "Completed copying table {0} to database {1}", destinationTableName, destination.Database))
            {
                using (IDbCommand cmd = source.CreateCommand())
                {
                    cmd.CommandText = sourceSQL;
                    try
                    {
                        source.Open();
                        destination.Open();
                        using (IDataReader rdr = cmd.ExecuteReader())
                        {
                            using (IDbCommand insertCmd = destination.CreateCommand())
                            {
                                string paramsSQL = String.Empty;

                                using (DataTable schemaTable = rdr.GetSchemaTable())
                                {
                                    //build the insert statement
                                    foreach (DataRow row in schemaTable.Rows)
                                    {
                                        if (paramsSQL.Length > 0)
                                            paramsSQL += ", ";
                                        paramsSQL += "@" + row["ColumnName"];

                                        IDbDataParameter param = insertCmd.CreateParameter();
                                        param.ParameterName = "@" + row["ColumnName"];
                                        param.SourceColumn = row["ColumnName"].ToString();

                                        if ((Type)row["DataType"] == typeof(System.DateTime))
                                        {
                                            param.DbType = DbType.DateTime;
                                        }

                                        insertCmd.Parameters.Add(param);
                                    }
                                }

                                string insertCommandText = string.Format("insert into {0} ( {1} ) values ( {2} )", destinationTableName, paramsSQL.Replace("@", String.Empty), paramsSQL);
                                insertCmd.CommandText = insertCommandText;

                                //kick off our worker processes...
                                Task[] processingTasks = new Task[ MaxDataParallelism ];
                                for (int curTaskIndex = 0; curTaskIndex < MaxDataParallelism; curTaskIndex++)
                                {
                                    processingTasks[curTaskIndex] = Task.Factory.StartNew(() => AsyncProcessCommand(pendingCommandQueue, destination.ConnectionString));
                                }

                                while (rdr.Read())
                                {
                                    foreach (IDbDataParameter param in insertCmd.Parameters)
                                    {
                                        object col = rdr[param.SourceColumn];
                                        param.Value = col ?? DBNull.Value;
                                    }

                                    pendingCommandQueue.Add(((SqlCommand)insertCmd).Clone());
                                }
                                pendingCommandQueue.CompleteAdding();

                                //let the queue drain and complete...
                                Task.WaitAll(processingTasks);
                            }
                        }
                    }
                    finally
                    {
                        destination.Close();
                        source.Close();
                    }
                }
            }
        }

        private static void AsyncProcessCommand(BlockingCollection<IDbCommand> pendingCommands, string connectionString)
        {
            try
            {
                using (var connection = new SqlConnection(connectionString))
                {
                    connection.Open();

                    foreach (var pendingCommand in pendingCommands.GetConsumingEnumerable())
                    {
                        using (pendingCommand)
                        {
                            pendingCommand.Connection = connection;
                            pendingCommand.ExecuteNonQuery();
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
            }
        }


        /// <summary>
        /// Truncate the specified table (delete all rows)
        /// </summary>
        /// <param name="target"></param>
        /// <param name="tableName"></param>
        public static void TruncateTable(IDbConnection target, string tableName)
        {
            using (IDbCommand cmd = target.CreateCommand())
            {
                cmd.CommandText = string.Format("DELETE {0};", tableName);
                cmd.CommandType = CommandType.Text;

                try
                {
                    target.Open();
                    cmd.ExecuteNonQuery();
                }
                finally
                {
                    target.Close();
                }
            }
        }
    }
}