c#-4.0 - 反应式扩展 + 异步与线程性能

Question

想知道这是否得到充分和有效的实施。我试图从中获得最快的速度，但我觉得我没有尽可能地实现这一点。这是代码片段。如果我调整 .Buffer() 调用并且只使用 1 （模拟单个线程），则需要更长的时间。如果我将缓冲区设置为 9（基于#of cores + 1），它会快得多。但是，我的印象是异步会比多线程更快......似乎在这里不起作用。

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Security.Cryptography;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Data;
using System.Data.Sql;
using System.Data.SqlClient;
using CryptSharp.Utility;
using CryptSharp; // CryptSharp Lib in NuGet
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Linq.Observαble;
using System.Reactive.PlatformServices;



namespace ConsoleApplication1
{
    internal class Program
    {
        private static void Main(string[] args)
        {
            const bool dispalyOutput = true;
            Console.WriteLine("Main thread id: {0}", Thread.CurrentThread.ManagedThreadId);
            // if the bugger is adjusted then it will create threads via .net TPL Parallel foreach
            var o = (from num in Enumerable.Range(1, 1000) select num).ToObservable(Scheduler.Default).Buffer(32);

            var w = Stopwatch.StartNew();

            o.SubscribeOn(Scheduler.Default).ObserveOn(Scheduler.Default).Subscribe(n =>
                        {
                            Parallel.ForEach(n, chunk =>
                            {
                            using (new SimultaneousDelegateCheck(false))
                            {
                                Thread.Sleep(0);
                                var crypto =  GetCrypto(chunk.ToString());
                                if (dispalyOutput)
                                    if (chunk%10 == 0)
                                        Console.WriteLine("Hash|{0} Tid:{1}: {2}", chunk,
                                                          Thread.CurrentThread.ManagedThreadId, crypto.Result);
                            }
                            });
                        },
                        (n) => Console.WriteLine("Error: {0}", n.Message),
                        () =>
                        {
                            Console.WriteLine("Done!");
                            Console.WriteLine("Total time: {0}", w.Elapsed);
                        });

            Console.WriteLine("Main thread finished in {0}", w.Elapsed);
            Console.WriteLine("Finished (called before anon delegate runs above)");
            Console.ReadLine();
        }

        static async Task<string> GetCrypto(string s)
        {
            //var crypto = Task.Run(() => s.CryptSharpSHA256Hash());  // requires CryptSharp lib from Nuget
            var crypto = Task.Run(() => s.Hash(new SHA256Managed()));
            //var crypto = Task.Run(() => s.Hash(new SHA256CryptoServiceProvider()));

            //var crypto = Task.Factory.StartNew(() => s.CryptSharpSHA256Hash());  // requires CryptSharp lib from Nuget
            //var crypto = Task.Run(() => s.Hash(new SHA256Managed()));
            //var crypto = Task.Run(() => s.Hash(new SHA256CryptoServiceProvider()));
            return await crypto;
        }
    }   

    public static class HashExtension
    {
        public static string CryptSharpScryptHash(this string input)
        {
            throw new NotImplementedException("don't do that");
        }

        public static string CryptSharpSHA256Hash(this string input)
        {
            return Crypter.Sha256.Crypt(input);
        }

        public static string Hash(this string input, HashAlgorithm algorithm = null)
        {
            if(algorithm==null)
                algorithm=new SHA256Managed();

            var inputBytes = Encoding.UTF8.GetBytes(input);
            // Combine salt and input bytes
            var salt = "hashsalt".GetBytes();
            var saltedInput = new Byte[salt.Length + inputBytes.Length];
            salt.CopyTo(saltedInput, 0);
            inputBytes.CopyTo(saltedInput, salt.Length);
            var hashedBytes = algorithm.ComputeHash(inputBytes);

            return BitConverter.ToString(hashedBytes).Replace("-", "");
            //return Convert.ToBase64String(hashedBytes);
        }

        public static byte[] GetBytes(this string str)
        {
            byte[] bytes = new byte[str.Length * sizeof(char)];
            System.Buffer.BlockCopy(str.ToCharArray(), 0, bytes, 0, bytes.Length);
            return bytes;
        }

        public static string GetString(this byte[] bytes)
        {
            char[] chars = new char[bytes.Length / sizeof(char)];
            System.Buffer.BlockCopy(bytes, 0, chars, 0, bytes.Length);
            return new string(chars);
        }
    }

    public class SimultaneousDelegateCheck : IDisposable
    {
        private static int _delegateCount = 0;
        public void Dispose()
        {
            Interlocked.Decrement(ref _delegateCount);
        }

        public SimultaneousDelegateCheck(bool display = true)
        {
            var usage = 0;
            if ((usage = Interlocked.Increment(ref _delegateCount)) == 1) return;
            if(display)
                Console.WriteLine("threads:{0}", _delegateCount);
        }
    }
}