有关对耗时很大循环进行并行化优化的探讨之二：多重循环任务的并发处理

【引】

在上一篇有关并行化的《有关对耗时很大循环进行并行化优化的探讨 之一：并发搜索的处理》博文中，我们看到，不使用并行语言，而使用基本的编程语言配合多线程可以进行并行化处理，本文进一步将并行化操作进行一个简单封装，并开始完成一个并行化的数据操作，以供大家参考。



【多重循环的操作】

有很多方法可以实现并行查询，必须并行化Linq就是很不错的途径，相比linq to entity，真正实现了性能的有效提升，但如果考虑在循环中进行复杂的业务处理，进行独立的并行化编程会更有效的帮助我们实现系统的优化。

下面我们以一个简单的操作为例，看看如何自己做一个并行化的程序：

在如下的例程中，我们自定义了一个继承于datatable的TLoopTable对象，并向该表格添加500×5条Tdata记录，很不幸的是，每条tdata记录需要10毫秒才能生成。使用该TLoopTable对象填充到datagridview的时间是39秒左右，

我们就以此应用开始进行我们的并行化优化。

class TLoopTable : DataTable

{

public const int CON = 500;

public TLoopTable()

{

this.Columns.Add("C1");

this.Columns.Add("D1");

this.Columns.Add("E1");

this.Columns.Add("F1");

this.Columns.Add("G1");

}

public void AddRows()

{

List<LoopD> plist = new List<LoopD>();

plist.Add(new LoopD("C", 1));

plist.Add(new LoopD("D", 2));

plist.Add(new LoopD("E", 3));

plist.Add(new LoopD("F", 4));

plist.Add(new LoopD("G", 5));

for (int i = 0; i < CON; i++)

{

TData [] tdl = new TData [5];

foreach (LoopD p in plist)

{

tdl[p.Col -1] = new TData(p.Col, i, p.Str) ;

}

this.Rows.Add(tdl);

}

}



class LoopD

{

public int Col;

public string Str;

public LoopD(string s, int c)

{

Col = c ;

Str = s ;

}

}

class TData

{

public int Row;

public int Col;

public string Data;

public TData(int col, int row,string fStr)

{

Thread.Sleep(10);

Row = row;

Col = col;

Data = fStr + row.ToString();





}

public override string ToString()

{

return Data;

}

}

}



首先，我们将上一篇文章所用的并行化方法进行一下封装如下：

　１. 通过定义一个委托，在指定我们在并行化操作内的具体输入输出，我这样定义委托，目的是达到的效果是：如果特定的操作需要循环强行退出（如发生异常），则通过taskRst=false实现，同时返回一个object类型值，以便外部程序了解中途退出的原因。

public delegate object MyParaLoopTaskHandler(object[] sender, int i, out bool taskRst);

class ParaLoop

{

private ManualResetEvent _ParaEvent;

private int _ExecedParaCount;

private object SynParaObj = new object();

private int ExecutedParaCount

{

get

{

lock (SynParaObj)

{

return _ExecedParaCount;

}

}

set

{

lock (SynParaObj)

{

_ExecedParaCount = value;

}

}

}

bool _IsParaFin;

private bool IsParaFin

{

get

{

lock (SynParaObj)

{

return _IsParaFin;

}

}

set

{

lock (SynParaObj)

{

_IsParaFin = value;

}

}

}

private int _TotalQty;

private object _ParaLoop_Return;

public object ParaCompute(MyParaLoopTaskHandler loopTask, object[] inArg, int loopQty)

{

_ParaEvent = new ManualResetEvent(false);

_IsParaFin = false;

_ExecedParaCount = 0;

_TotalQty = loopQty;

for (int i = 0; i < _TotalQty; i++)

{

MyParaThread u = new MyParaThread();

u.EndTaskCallBack +=new EndTaskHandler(u_EndTaskCallBack);

u.BeginInvoke(loopTask, inArg, i);

}

//使用waitone进行阻塞，仅仅是保证外部调用时使用效果上与顺序执行相同，但如果希望使用异步处理，这里可以完全不用阻塞，而通过_EndTaskCallBack返回一个事件即可

_ParaEvent.WaitOne();

return _ParaLoop_Return;

}

void u_EndTaskCallBack(bool taskRst, object retVal)

{

if (IsParaFin)

return;

if (!taskRst)

{

_ParaLoop_Return = retVal;

_ParaEvent.Set();

return;

}

lock (SynParaObj)

{

ExecutedParaCount++;

if (ExecutedParaCount >= _TotalQty)

{

IsParaFin = true;

_ParaEvent.Set();

}

}

}

}

internal delegate void EndTaskHandler(bool taskRst,object retVal);

class MyParaThread

{

internal event EndTaskHandler EndTaskCallBack;

public void BeginInvoke(MyParaLoopTaskHandler thdTask, object[] sender,int i)

{

Thread p = new Thread(new ParameterizedThreadStart(DoThreadTask), 262144);

ParaTaskContent t = new ParaTaskContent(thdTask, sender,i);

p.Start(t);

}

private void DoThreadTask(object args)

{

ParaTaskContent c = args as ParaTaskContent;

bool rs = true;

object retval= c.Invoke( out rs );

if (EndTaskCallBack != null)

{

EndTaskCallBack(rs, retval);

}

}

class ParaTaskContent

{

public MyParaLoopTaskHandler _Task;

object[] _Sender;

int _Loop_i;

public ParaTaskContent(MyParaLoopTaskHandler thdTask, object[] sender,int loop_i)

{

_Task = thdTask;

_Sender = sender;

_Loop_i = loop_i;

}

public object Invoke(out bool rst)

{

return _Task.Invoke(_Sender, _Loop_i, out rst);

}

}

}



经过封装以后，外部程序只需要如此调用即可实现并行化操作：

1.定义一个要处理的委托：

MyParaLoopTaskHandler k = new MyParaLoopTaskHandler(PerformParaCompute);

2. 创建一个并行化对象

ParaLoop t = new ParaLoop();

3. 指定需要输入的内容

object[] inargs = new object[2];

inargs[0] = plist;

inargs[1] = rlist;

4. 执行并行化即可

t.ParaCompute(k, inargs,TLoopTable .CON );

而并行化的处理结果可以通过委托方法中定义的操作实现。在本例中，我们每次并行化操作是生成一行（5项TData记录）记录，但我们需要一个途径使并行化生成的行数据与自定义表格容器的行向对应，否则结果可能是显示顺序不是0，1，2，3，...，而是0,3,2,4,1...

为了实现此目的，我们就先生成空白的列表数据，同时创建一个有行标记的数据字典，生成完数据后，再一次填充入table中。以下就是实现的例程：

public void AddRows()

{

List<LoopD> plist = new List<LoopD>();

plist.Add(new LoopD("C", 1));

plist.Add(new LoopD("D", 2));

plist.Add(new LoopD("E", 3));

plist.Add(new LoopD("F", 4));

plist.Add(new LoopD("G", 5));

//生成空白表

for (int i=0;i<CON ;i++)

{

this.Rows.Add(new object [5]);

}

ParaLoop t = new ParaLoop();

Dictionary<int, TData[]> rlist = new Dictionary<int, TData[]>(); // 用于保证并行化数据的顺序与表格位置一致的数据字典

object[] inargs = new object[2];

inargs[0] = plist; //传递给并行处理程序 生成没行数据对应位置信息所需要的子循环体

inargs[1] = rlist; // 传递给并行处理程序位置数据字典

MyParaLoopTaskHandler k = new MyParaLoopTaskHandler(PerformParaCompute);

t.ParaCompute(k, inargs,TLoopTable .CON );

//并行化完成后，将数据字典填充到表格中。

foreach (int row in rlist.Keys)

{

foreach (TData d in rlist[row])

{

this.Rows[row][d.Col-1] = d;

}

}

}

//并行处理数据的执行方法

private object PerformParaCompute(object[] sender, int i, out bool taskRst)

{

object[] k = sender as object[];

//提取到输入数据，并进行转换

List<LoopD> loopsrc = k[0] as List<LoopD>;

Dictionary<int, TData[]> rlist = k[1] as Dictionary<int, TData[]>;

TData[] r = new TData[5];

foreach (LoopD l in loopsrc)

{

r[l.Col - 1] = new TData(l.Col, i, l.Str);

}

lock (rlist)

{

rlist.Add(i, r);

}

taskRst = true;

return null;

}



经过并行化改造前后的执行效率如下表

执行效率比较

	串行执行	并行化执行
执行时间	39.072秒	2.172秒
CPU占用率	<1%	68%

可以看到，通过并行化改造，系统效率提升了20倍。哇噢。