并发MD5计算方法

MD5与SHA算法一样，利用他们可以计算某段数据的唯一hash值，常用做校验码。而MD5比SHA算法性能高。在我参加的一个项目中，主要用MD5码值来去重，因此对计算性能要求较高。网上有对MD5算法并行化方法，能保证计算结果与线性计算法一样。由于我只需要唯一标记，并不要求它与线性MD5计算结果一样，所以，我通过分片多线程计算的方法实现提速，也充分利用了多核CPU的并行计算优势。

依据：

B  = （b1,b2,b3,b4...bn）--->(m1,m2,m3...mn)----+--->M---->(md5) result = 16 bytes

C#代码：

class SafeQueue<T>
{
   // A queue that is protected by Monitor. 
   private Queue<T> m_inputQueue = new Queue<T>();

   // Lock the queue and add an element. 
   public void Enqueue(T qValue)
   {
      // Request the lock, and block until it is obtained.
      Monitor.Enter(m_inputQueue);
      try
      {
         // When the lock is obtained, add an element.
         m_inputQueue.Enqueue(qValue);
      }
      finally
      {
         // Ensure that the lock is released.
         Monitor.Exit(m_inputQueue);
      }
   }

   // Try to add an element to the queue: Add the element to the queue  
   // only if the lock is immediately available. 
   public bool TryEnqueue(T qValue)
   {
      // Request the lock. 
      if (Monitor.TryEnter(m_inputQueue))
      {
         try
         {
            m_inputQueue.Enqueue(qValue);
         }
         finally
         {
            // Ensure that the lock is released.
            Monitor.Exit(m_inputQueue);
         }
         return true;
      }
      else
      {
         return false;
      }
   }

   // Try to add an element to the queue: Add the element to the queue  
   // only if the lock becomes available during the specified time 
   // interval. 
   public bool TryEnqueue(T qValue, int waitTime)
   {
      // Request the lock. 
      if (Monitor.TryEnter(m_inputQueue, waitTime))
      {
         try
         {
            m_inputQueue.Enqueue(qValue);
         }
         finally
         {
            // Ensure that the lock is released.
            Monitor.Exit(m_inputQueue);
         }
         return true;
      }
      else
      {
         return false;
      }
   }

   // Lock the queue and dequeue an element. 
   public T Dequeue()
   {
      T retval;

      // Request the lock, and block until it is obtained.
      Monitor.Enter(m_inputQueue);
      try
      {
         // When the lock is obtained, dequeue an element.
         retval = m_inputQueue.Dequeue();
      }
      finally
      {
         // Ensure that the lock is released.
         Monitor.Exit(m_inputQueue);
      }

      return retval;
   }

   // Delete all elements that equal the given object. 
   public int Remove(T qValue)
   {
      int removedCt = 0;

      // Wait until the lock is available and lock the queue.
      Monitor.Enter(m_inputQueue);
      try
      {
         int counter = m_inputQueue.Count;
         while (counter > 0)
            // Check each element.
         {
            T elem = m_inputQueue.Dequeue();
            if (!elem.Equals(qValue))
            {
               m_inputQueue.Enqueue(elem);
            }
            else
            {
               // Keep a count of items removed.
               removedCt += 1;
            }
            counter = counter - 1;
         }
      }
      finally
      {
         // Ensure that the lock is released.
         Monitor.Exit(m_inputQueue);
      }

      return removedCt;
   }

   // Print all queue elements. 
   public string PrintAllElements()
   {
      StringBuilder output = new StringBuilder();

      // Lock the queue.
      Monitor.Enter(m_inputQueue);
      try
      {
         foreach( T elem in m_inputQueue )
         {
            // Print the next element.
            output.AppendLine(elem.ToString());
         }
      }
      finally
      {
         // Ensure that the lock is released.
         Monitor.Exit(m_inputQueue);
      }

      return output.ToString();
   }
}


//任务节点
public class TaskItem
{
   public int idx_;
   public byte[] data_;
   public TaskItem( int idx, byte[] data )
   {
      this.idx_ = idx;
      this.data_ = data;
   }
}
//工作线程
public class WorkThread
{
    SafeQueue<TaskItem>  packet_;
     AutoResetEvent    notifyEvt_;
     ManualResetEvent     finishEvt_;
     Thread            thread_;
      HashTable         result_;
      
     public WorkThread()
     {
        ....
        thread_ = new Thread( new ThreadStart( this.DoWork ) );
     }
     public void Start()
     {
          thread_.Start();
     }
     //主线程调用：异步添加任务节点
     public void AddDataPacket( TaskItem data )
     {
         packet_.Enqueue( data );
         notifyEvt_.Set();
         
     }
     //主线程调用：等待收集计算结果
     public Hashtable Finish()
     {
            Hashtable rs = new Hashtable( result_ );
           finishEvt_.WaitOne();
           finishEvt_.Reset();
           result_.clear();
           return rs;
     }
     //线程执行函数
     public void DoWork()
     {
           TaskItem  var;
           while(true)
           {
               notifyEvt_.WaitOne();
               while(true)
               {
                   if( packet_.Count > 0 )
                   {
                        //这里取出值错误
                        var = packet_.Dequeue();
                   }
                   if( var == null )
                       break;
                    //task finish
                   if( var.Idx == -1 )
                   {
                       ....
                       this.finishEvt_.Set();
                   }
               }
           }
     }
}

public class Md5Hasher
{
   public void main()
   {
         WorkThread[] test=new WorkThread[5];

         for( int i=0; i<100; ++i )
         {
                //这里输入数据
               test[i%5].AddDataPacket( ... );
         }
   }
}