RabbitMQ + .NET C#：消息发布 + 消费的低性能和可变性能

Question

我正在尝试使用 RabbitMQ + .NET C# 提高发布和消费消息的性能。

我在下面定义了相对“大”的 xml 文件（作为嵌入式资源）：

<?xml version="1.0" encoding="UTF-8"?>
<Document>
  <CstmrCdtTrfInitn>
    <GrpHdr>
      <MsgId>ABC/090928/CCT001</MsgId>
      <CreDtTm>2009-09-28T14:07:00</CreDtTm>
      <NbOfTxs>3</NbOfTxs>
      <CtrlSum>11500000</CtrlSum>
      <InitgPty>
        <Nm>ABC Corporation</Nm>
        <PstlAdr>
          <StrtNm>Times Square</StrtNm>
          <BldgNb>7</BldgNb>
          <PstCd>NY 10036</PstCd>
          <TwnNm>New York</TwnNm>
          <Ctry>US</Ctry>
        </PstlAdr>
      </InitgPty>
    </GrpHdr>
    <PmtInf>
      <PmtInfId>ABC/086</PmtInfId>
      <PmtMtd>TRF</PmtMtd>
      <BtchBookg>false</BtchBookg>
      <ReqdExctnDt>2009-09-29</ReqdExctnDt>
      <Dbtr>
        <Nm>ABC Corporation</Nm>
        <PstlAdr>
          <StrtNm>Times Square</StrtNm>
          <BldgNb>7</BldgNb>
          <PstCd>NY 10036</PstCd>
          <TwnNm>New York</TwnNm>
          <Ctry>US</Ctry>
        </PstlAdr>
      </Dbtr>
      <DbtrAcct>
        <Id>
          <Othr>
            <Id>00125574999</Id>
          </Othr>
        </Id>
      </DbtrAcct>
      <DbtrAgt>
        <FinInstnId>
          <BIC>BBBBUS33</BIC>
        </FinInstnId>
      </DbtrAgt>
      <CdtTrfTxInf>
        <PmtId>
          <InstrId>ABC/090928/CCT001/01</InstrId>
          <EndToEndId>ABC/4562/2009-09-08</EndToEndId>
        </PmtId>
        <Amt>
          <InstdAmt Ccy="JPY">10000000</InstdAmt>
        </Amt>
        <ChrgBr>SHAR</ChrgBr>
        <CdtrAgt>
          <FinInstnId>
            <BIC>AAAAGB2L</BIC>
          </FinInstnId>
        </CdtrAgt>
        <Cdtr>
          <Nm>DEF Electronics</Nm>
          <PstlAdr>
            <AdrLine>Corn Exchange 5th Floor</AdrLine>
            <AdrLine>Mark Lane 55</AdrLine>
            <AdrLine>EC3R7NE London</AdrLine>
            <AdrLine>GB</AdrLine>
          </PstlAdr>
        </Cdtr>
        <CdtrAcct>
          <Id>
            <Othr>
              <Id>23683707994125</Id>
            </Othr>
          </Id>
        </CdtrAcct>
        <Purp>
          <Cd>CINV</Cd>
        </Purp>
        <RmtInf>
          <Strd>
            <RfrdDocInf>
              <Nb>4562</Nb>
              <RltdDt>2009-09-08</RltdDt>
            </RfrdDocInf>
          </Strd>
        </RmtInf>
      </CdtTrfTxInf>
      <CdtTrfTxInf>
        <PmtId>
          <InstrId>ABC/090628/CCT001/2</InstrId>
          <EndToEndId>ABC/ABC-13679/2009-09-15</EndToEndId>
        </PmtId>
        <Amt>
          <InstdAmt Ccy="EUR">500000</InstdAmt>
        </Amt>
        <ChrgBr>CRED</ChrgBr>
        <CdtrAgt>
          <FinInstnId>
            <BIC>DDDDBEBB</BIC>
          </FinInstnId>
        </CdtrAgt>
        <Cdtr>
          <Nm>GHI Semiconductors</Nm>
          <PstlAdr>
            <StrtNm>Avenue Brugmann</StrtNm>
            <BldgNb>415</BldgNb>
            <PstCd>1180</PstCd>
            <TwnNm>Brussels</TwnNm>
          </PstlAdr>
        </Cdtr>
        <CdtrAcct>
          <Id>
            <IBAN>BE30001216371411</IBAN>
          </Id>
        </CdtrAcct>
        <InstrForCdtrAgt>
          <Cd>PHOB</Cd>
          <InstrInf>+32/2/2222222</InstrInf>
        </InstrForCdtrAgt>
        <Purp>
          <Cd>GDDS</Cd>
        </Purp>
        <RmtInf>
          <Strd>
            <RfrdDocInf>
              <Tp>
                <CdOrPrtry>
                  <Cd>CINV</Cd>
                </CdOrPrtry>
              </Tp>
              <Nb>ABC-13679</Nb>
              <RltdDt>2009-09-15</RltdDt>
            </RfrdDocInf>
          </Strd>
        </RmtInf>
      </CdtTrfTxInf>
      <CdtTrfTxInf>
        <PmtId>
          <InstrId>ABC/090928/CCT001/3</InstrId>
          <EndToEndId>ABC/987-AC/2009-09-27</EndToEndId>
        </PmtId>
        <Amt>
          <InstdAmt Ccy="USD">1000000</InstdAmt>
        </Amt>
        <ChrgBr>SHAR</ChrgBr>
        <CdtrAgt>
          <FinInstnId>
            <BIC>BBBBUS66</BIC>
          </FinInstnId>
        </CdtrAgt>
        <Cdtr>
          <Nm>ABC Corporation</Nm>
          <PstlAdr>
            <Dept>Treasury department</Dept>
            <StrtNm>Bush Street</StrtNm>
            <BldgNb>13</BldgNb>
            <PstCd>CA 94108</PstCd>
            <TwnNm>San Francisco</TwnNm>
            <Ctry>US</Ctry>
          </PstlAdr>
        </Cdtr>
        <CdtrAcct>
          <Id>
            <Othr>
              <Id>4895623</Id>
            </Othr>
          </Id>
        </CdtrAcct>
        <Purp>
          <Cd>INTC</Cd>
        </Purp>
        <RmtInf>
          <Strd>
            <RfrdDocInf>
              <Tp>
                <CdOrPrtry>
                  <Cd>CINV</Cd>
                </CdOrPrtry>
              </Tp>
              <Nb>987-AC</Nb>
              <RltdDt>2009-09-27</RltdDt>
            </RfrdDocInf>
          </Strd>
        </RmtInf>
      </CdtTrfTxInf>
    </PmtInf>
  </CstmrCdtTrfInitn>
</Document>

然后我将该文件发布到同一个队列大约 500,000 次，发布和消费发生在下面的同一个 .NET Core C# 程序中：

.NET C# 代码：

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;

using RabbitMQ.Client;
using RabbitMQ.Client.Events;


namespace CSharpPlayground
{
    public static class EmbeddedResource
    {
        public static string Load(string fileName)
        {
            var executingAssembly = Assembly.GetExecutingAssembly();
            var resource = $"{executingAssembly.ManifestModule.Name.Replace(".dll", string.Empty)}.{fileName}";
            using var stream = executingAssembly.GetManifestResourceStream(resource);
            using var streamReader = new StreamReader(stream!);
            return streamReader.ReadToEnd();
        }
    }

    public static class CheapExtensions
    {
        public static void ForEach<T>(this IEnumerable<T> source, Action<T> action)
        {
            foreach(var item in source)
            {
                action(item);
            }
        }
        public static void ForEach<T>(this IEnumerable<T> source) =>
            source.ForEach(item => {});
    }

    public static class Program
    {
        private static readonly string XmlStuff = EmbeddedResource.Load("XmlStuff.xml");

        public static void Main()
        {
            const string queueName = "hello";
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                DispatchConsumersAsync = true
            };
            using var connection = factory.CreateConnection();
            using var queueDeclareChannel = connection.CreateModel();

            queueDeclareChannel.QueueDeclare(
                queue: queueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: null);

            using var consumerChannel = connection.CreateModel();
            var consumer = new AsyncEventingBasicConsumer(consumerChannel);
            consumer.Received += async (sender, eventArgs) =>
            {
                var receivedBody = eventArgs.Body;
                var receivedMessage = Encoding.UTF8.GetString(receivedBody.ToArray());
                //Console.WriteLine($"[x] Received stuff");
            };
            consumerChannel.BasicConsume(
                queue: queueName,
                autoAck: true,
                consumer: consumer);

            var messageToSend = XmlStuff;
            var bodyToSend = new ReadOnlyMemory<byte>(Encoding.UTF8.GetBytes(messageToSend));

            Enumerable
                .Repeat((messageToSend, bodyToSend), 500_000)
                .AsParallel()
                .Select((data, index) =>
                {
                    var (message, body) = data;
                    using var publishChannel = connection.CreateModel();
                    var basicProperties = publishChannel.CreateBasicProperties();
                    publishChannel.BasicPublish(
                        exchange: "",
                        routingKey: queueName,
                        mandatory: false,
                        basicProperties: basicProperties,
                        body: body);
                    //Console.WriteLine($"[x] Sent stuff"
                    return index;
                }).ForEach();

            Console.ReadKey();
        }
    }
}

我已尝试尽可能删除 Console.WriteLine 调用以规避将某些内容推送到标准输出的瓶颈。

事实证明，性能似乎相当低，我想知道我的性能（就消息率而言）是否“正常”：

Answer 1

我不是 RabbitMQ 专家，但我知道为要发布的每条消息创建一个通道 (CreateModel()) 是 RabbitMQ 反模式，会损害性能。这是因为每个通道创建都需要到代理的网络往返。这种方法也有让经纪人筋疲力尽的风险。

不幸的是，很多 RabbitMQ 入门博客都是这样开始的，可能是因为它简化了问题。可以这么说，问题在于通道不是线程安全的。

如果您想在 RabbitMQ 之上编写自己的消息传递框架，那么我将从单个发布通道开始并使用锁来同步它。

如果您想构建一个真正的应用程序，避免重新发明轮子和许多令人头疼的问题，我会看看一些得到良好支持的开源框架，例如 EasyNetQ 或 MassTransit。