并行 LINQ (PLINQ) 是 LINQ to Objects 的并行实现。 PLINQ 实现完整的 LINQ 标准查询运算符集作为 T:System.Linq 命名空间的扩展方法，并具有用于并行运算的其他运算符。 PLINQ 将 LINQ 语法的简洁和可靠性与并行编程的强大功能结合在一起。 就像面向任务并行库的代码一样，PLINQ 查询会根据主计算机的能力按比例调整并发程度。
在许多情况下，PLINQ 可通过更有效地使用主计算机上的所有可用内核来显著提高 LINQ to Objects 查询的速度。 这一性能提升将使桌面具备高性能计算能力。
为了实现加速，PLINQ 查询必须具有足够的适合并行工作来弥补开销。 工作可表示为每个委托的计算开销与源集合中元素数量的乘积。 假定某个操作可并行化，则它的计算开销越高，加速的可能性就越大。 例如，如果某个函数执行花费的时间为 1 毫秒，则针对 1000 个元素进行的顺序查询将花费 1 秒来执行该操作，而在四核计算机上进行的并行查询可能只花费 250 毫秒。 这样就产生了 750 毫秒的加速。 如果该函数对于每个元素需要花费 1 秒来执行，则加速将为 750 秒。如果委托的开销很大，则对于源集合中的很少几个项，PLINQ 可能会提供明显的加速。 相反，包含无关紧要委托的小型源集合通常不适合于 PLINQ。——MSDN
1.普通示例
usingSystem;
usingSystem.Collections;
usingSystem.Collections.Generic;
usingSystem.Diagnostics;
usingSystem.Linq;
usingSystem.Threading.Tasks;
namespacePLinq
{
classPRogram
{
staticIList<Person> _persons =newList<Person>();
staticvoidMain(string[] args)
{
try
{
CreateTextDB();
PLINQ_Where();
}
catch(Exception ex)
{
Console.WriteLine(string.Format("Exception Message:{0}", ex.Message.Trim()));
}
finally
{
Console.ReadLine();
}
}
/// <summary>
/// 筛选
/// </summary>
privatestaticvoidPLINQ_Where()
{
Stopwatch _wacth =newStopwatch();
_wacth.Start();
IList<Person> _personWhereNormal = _persons.Where(p => p.Age > 10 && p.Age < 50).ToList();
_wacth.Stop();
Console.WriteLine(string.Format("Normal LINQ Where Cost Time:{0}", _wacth.ElapsedMilliseconds));
_wacth.Restart();
//WithDegreeOfParallelism
IList<Person> _personWhereParallel = _persons.asparallel<Person>().Where(p => p.Age > 10 && p.Age < 50).ToList();
_wacth.Stop();
Console.WriteLine(string.Format("PLINQ Where Cost Time:{0}", _wacth.ElapsedMilliseconds));
}
/// <summary>
/// 创建测试数据
/// </summary>
privatestaticvoidCreateTextDB()
{
Stopwatch _wacth =newStopwatch();
_wacth.Start();
Parallel.For(0, 10000000, (i, loopstatus) =>
{
Person _person =newPerson();
_person.FristName ="Yan";
_person.LastName =string.Format("Zhiwei {0}", i);
_person.RegisterTime = DateTime.Now;
_person.Age = i;
lock(((ICollection)_persons).SyncRoot)
{
_persons.Add(_person);
}
});
_wacth.Stop();
Console.WriteLine(string.Format("Create TextDB Cost Time:{0},Count:{1}", _wacth.ElapsedMilliseconds, _persons.Count));
}
}
classPerson
{
publicstringFristName { get; set; }
publicstringLastName { get; set; }
publicintAge { get; set; }
publicDateTime RegisterTime { get; set; }
}
}
.csharpcode, .csharpcode pre
{
font-size: small;
color: black;
font-family: consolas, "Courier New", courier, monospace;
background-color: #ffffff;
/*white-space: pre;*/
}
.csharpcode pre { margin: 0em; }
.csharpcode .rem { color: #008000; }
.csharpcode .kwrd { color: #0000ff; }
.csharpcode .str { color: #006080; }
.csharpcode .op { color: #0000c0; }
.csharpcode .preproc { color: #cc6633; }
.csharpcode .asp { background-color: #ffff00; }
.csharpcode .html { color: #800000; }
.csharpcode .attr { color: #ff0000; }
.csharpcode .alt
{
background-color: #f4f4f4;
width: 100%;
margin: 0em;
}
.csharpcode .lnum { color: #606060; }代码效果
2.异常处理
usingSystem;usingSystem.Collections;usingSystem.Collections.Generic;usingSystem.Diagnostics;usingSystem.Linq;usingSystem.Threading.Tasks;namespacePLinq
{classProgram
{staticIList<Person> _persons =newList<Person>();staticvoidMain(string[] args)
{try{
CreateTextDB();
PINQ_Exception();
}catch(Exception ex)
{
Console.WriteLine(string.Format("Exception Message:{0}", ex.Message.Trim()));
}finally{
Console.ReadLine();
}
}privatestaticvoidPINQ_Exception()
{/** 这样处理，当出现异常的时候不会影响下次foralll遍历*/Func<int,bool> isTure = (age) =>
{try{if(age > 1989222)thrownewException("PLINQ Text");
Console.WriteLine(age);returntrue;
}catch(Exception ex)
{
Console.WriteLine(ex.Message);returnfalse;
}
};
_persons.AsParallel<Person>().ForAll(p => isTure(p.Age));
}/// <summary>/// 创建测试数据/// </summary>privatestaticvoidCreateTextDB()
{
Stopwatch _wacth =newStopwatch();
_wacth.Start();
Parallel.For(0, 10000000, (i, loopstatus) =>
{
Person _person =newPerson();
_person.FristName ="Yan";
_person.LastName =string.Format("Zhiwei {0}", i);
_person.RegisterTime = DateTime.Now;
_person.Age = i;lock(((ICollection)_persons).SyncRoot)
{
_persons.Add(_person);
}
});
_wacth.Stop();
Console.WriteLine(string.Format("Create TextDB Cost Time:{0},Count:{1}", _wacth.ElapsedMilliseconds, _persons.Count));
}
}classPerson
{publicstringFristName { get; set; }publicstringLastName { get; set; }publicintAge { get; set; }publicDateTime RegisterTime { get; set; }
}
}
.csharpcode, .csharpcode pre
{
font-size: small;
color: black;
font-family: consolas, "Courier New", courier, monospace;
background-color: #ffffff;
/*white-space: pre;*/
}
.csharpcode pre { margin: 0em; }
.csharpcode .rem { color: #008000; }
.csharpcode .kwrd { color: #0000ff; }
.csharpcode .str { color: #006080; }
.csharpcode .op { color: #0000c0; }
.csharpcode .preproc { color: #cc6633; }
.csharpcode .asp { background-color: #ffff00; }
.csharpcode .html { color: #800000; }
.csharpcode .attr { color: #ff0000; }
.csharpcode .alt
{
background-color: #f4f4f4;
width: 100%;
margin: 0em;
}
.csharpcode .lnum { color: #606060; }代码效果
3.取消 PLINQ 查询
usingSystem;usingSystem.Collections;usingSystem.Collections.Generic;usingSystem.Diagnostics;usingSystem.Linq;usingSystem.Threading;usingSystem.Threading.Tasks;namespacePLinq
{classProgram
{staticIList<Person> _persons =newList<Person>();staticvoidMain(string[] args)
{try{
CreateTextDB();
PINQ_Cancellation();
}catch(Exception ex)
{
Console.WriteLine(string.Format("Exception Message:{0}", ex.Message.Trim()));
}finally{
Console.ReadLine();
}
}privatestaticvoidPINQ_Cancellation()
{try{/** 当您在用户代码中处理取消时，不必在查询定义中使用 WithCancellation<TSource>。 但是，我们建议您这样做，原因是 WithCancellation<TSource> 对查询性能没有影响，* 并且它使取消能够由查询运算符和用户代码进行处理。 为了确保系统响应能力，我们建议您大约每毫秒检查是否存在取消一次；* 不过，任何 10 毫秒以下的期间都被视为可接受。 此频率对代码的性能应没有负面影响。 ——MSDN*/CancellationTokenSource _cancel =newCancellationTokenSource();Func<int,bool> isTure = (age) =>
{if(age > 1989222)
_cancel.Cancel();
Console.WriteLine(age);returntrue;
};
_persons.AsParallel<Person>().WithCancellation(_cancel.Token).ForAll(p => isTure(p.Age));
}catch(OperationCanceledException ex)
{
Console.WriteLine(string.Format("OperationCanceledException Message:{0}", ex.Message));
}
}/// <summary>/// 创建测试数据/// </summary>privatestaticvoidCreateTextDB()
{
Stopwatch _wacth =newStopwatch();
_wacth.Start();
Parallel.For(0, 10000000, (i, loopstatus) =>
{
Person _person =newPerson();
_person.FristName ="Yan";
_person.LastName =string.Format("Zhiwei {0}", i);
_person.RegisterTime = DateTime.Now;
_person.Age = i;lock(((ICollection)_persons).SyncRoot)
{
_persons.Add(_person);
}
});
_wacth.Stop();
Console.WriteLine(string.Format("Create TextDB Cost Time:{0},Count:{1}", _wacth.ElapsedMilliseconds, _persons.Count));
}
}classPerson
{publicstringFristName { get; set; }publicstringLastName { get; set; }publicintAge { get; set; }publicDateTime RegisterTime { get; set; }
}
}
.csharpcode, .csharpcode pre
{
font-size: small;
color: black;
font-family: consolas, "Courier New", courier, monospace;
background-color: #ffffff;
/*white-space: pre;*/
}
.csharpcode pre { margin: 0em; }
.csharpcode .rem { color: #008000; }
.csharpcode .kwrd { color: #0000ff; }
.csharpcode .str { color: #006080; }
.csharpcode .op { color: #0000c0; }
.csharpcode .preproc { color: #cc6633; }
.csharpcode .asp { background-color: #ffff00; }
.csharpcode .html { color: #800000; }
.csharpcode .attr { color: #ff0000; }
.csharpcode .alt
{
background-color: #f4f4f4;
width: 100%;
margin: 0em;
}
.csharpcode .lnum { color: #606060; }代码效果