protobuf.net 通过多线程

Question

我想使用 protobuf.net 反序列化大量数据，但我发现它无法通过多线程提高吞吐量。

我的测试场景：

1. 单线程，CPU负载为25%（1/4 cpu资源）
2. 单线程，4 个进程，cpu 负载为 9x %（4/4 cpu 资源）
3. 4线程，1进程，cpu负载为30%~60%

也就是说，protobuf 在多线程下无法充分利用 cpu 资源。

这是我的代码

    private static void DeSerialize()
    {
        while (true)
        { 
            Dictionary<string, byte[]>  cache ;
            if (queue.TryDequeue(out cache))
            {
                foreach (byte[] unit in cache.Values)
                {
                    using (MemoryStream stream = new MemoryStream(unit))
                    {
                        CommonUtil.DeSerializeBuf<User>(stream);
                    }
                }
            }
            else break;
        }
    }

    private static void DeSerializeThread()
    {
        for (int i = 0; i < 4; i++)
        {
            Thread a = new Thread(DeSerialize);
            a.Start();
        }
    }

protobuf.net 如何通过多线程充分利用多个 CPU 资源？在我的情况下，多进程是不可接受的。

我的并行测试代码：

        var dic = new Dictionary<string, byte[]>();
        for (int i = 0; i < 10000000; i++)
        {
            MemoryStream stream = new MemoryStream();
            ProtoBuf.Serializer.Serialize<string>(stream, "some value which is awesome" + i);
            byte[] buffer = stream.ToArray();
            dic.Add("key" + i, buffer);
        }

        var watch = new Stopwatch();
        watch.Restart();
        Console.Write("start parallel..");

        var result = dic.AsParallel().Select(p => ProtoBuf.Serializer.Deserialize<string>(new MemoryStream(p.Value))).ToList();
        var p1 = watch.ElapsedMilliseconds;
        Console.WriteLine("end parallel " + p1);

        watch.Restart();
        Console.Write("start sequential..");

        var result2 = dic.Select(p => CommonUtil.DeSerializeBuf<string>(new MemoryStream(p.Value))).ToList();
        var p2 = watch.ElapsedMilliseconds;
        Console.WriteLine("end parallel " + p2);

谢谢。

Answer 1

我可以给你一个简短的例子来说明如何使用 Linq 作为并行，也许这会有所帮助

var dic = new Dictionary<string, byte[]>();
for (int i = 0; i < 1000000; i++)
{
    dic.Add("key" + i, Serialize<string>("some value which is awesome" + i));
}

var watch = new Stopwatch();
watch.Restart();
Console.WriteLine("start parallel");

var result = dic.AsParallel().Select(p => Deserialize<string>(p.Value)).ToList();
var p1 = watch.ElapsedMilliseconds;

watch.Restart();
Console.WriteLine("start sequential");

var result2 = dic.Select(p => Deserialize<string>(p.Value)).ToList();
var p2 = watch.ElapsedMilliseconds;

在我的机器上并行执行的速度大约快 3 倍。而且处理器利用率是不同的