并发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( ... );
}
}
}