【发布时间】：2021-07-19 18:10:18
【问题描述】：
我想知道是否有人可以帮助解决以下情况：
我无法使用用 C# 编写并使用 .Net core 5.0 的 RabbitMQ Publisher 解决内存泄漏。
这是 csproj 文件：
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework>
<RuntimeIdentifier>win-x64</RuntimeIdentifier>
</PropertyGroup>
...
</Project>
我有一个在虚拟机中运行的 .net 控制台应用程序，它通过 API（注册的 64 位 dll 并作为 COM 引用引用）连接到服务器，从该服务器获取信息，然后尝试将此信息发布到位于 AWS 云上的 RabbitMQ 机器（该 RMQ 实例具有多个节点的负载均衡器）。
通过以下方式访问代码中的API：
private void SetUpApi () {
Utils.log.Info (_api.SetAPIOptions ("<CONNECTIONOPTIONS><CALCULATED_PRICES Enabled='true' MaximumDepth='4'/></CONNECTIONOPTIONS>"));
_api.OnServerConnect += OnServerConnect;
_api.OnServerDisconnect += OnServerDisconnect;
_api.OnNewData += OnNewData;
}
private void OnNewData(string strXML){
try{
if (strXML.Contains("<ORDER")){
ParseXMLAnswer(strXML, "OnNewData ()");
}
}
catch (Exception ex) {
if (ex.InnerException is AlreadyClosedException || ex.InnerException is BrokerUnreachableException)
Utils.log.Error("OnNewData () RabbitMQ.Client.Exceptions.AlreadyClosedException ");
else
Utils.printException("OnNewData ()", ex);
}
}
private void ParseXMLAnswer(string strOutputXML, string caller) {
XmlDocument doc = new XmlDocument();
doc.LoadXml(strOutputXML);
string jsonText = JsonConvert.SerializeXmlNode(doc);
var o = JObject.Parse(jsonText);
if (o["APIDATA"]["ORDER"].Type is JTokenType.Object){
JObject order = (JObject)o["APIDATA"]["ORDER"];
SendOrderToRMQ(order);
}
else if (o["APIDATA"]["ORDER"].Type is JTokenType.Array){
JArray orders = (JArray)o["APIDATA"]["ORDER"];
foreach (var item in orders.Children()){
SendOrderToRMQ((JObject)item);
}
}
doc = null;
GC.Collect();
GC.WaitForPendingFinalizers();
}
private void SendOrderToRMQ (JObject order){
JObject instrSpeciefier = (JObject) order["INSTSPECIFIER"];
var firstSeqID = instrSpeciefier.GetValue("@FirstSequenceID").ToString();
var firstSeqItemID = instrSpeciefier.GetValue("@FirstSequenceItemID").ToString();
if (sequenceItemsHashed.ContainsKey(firstSeqID) &&
sequenceItemsHashed[firstSeqID].Contains(firstSeqItemID)){
string itemName = Utils.ReplaceSensitiveCharacthers(instrSpeciefier.GetValue("@FirstSequenceItemName").ToString());
string instrumentName = Utils.ReplaceSensitiveCharacthers(instrSpeciefier.GetValue("@InstName").ToString());
int index = sequenceItemsHashed[firstSeqID].IndexOf(firstSeqItemID) + 1;
var binding = instrumentName + "." + sequencesFromInstruments[firstSeqID] + "." + itemName + "." + index;
serviceInstance1.Publish(
order.ToString(),
_exchangeName,
"",
binding);
}
order = null;
instrSpeciefier = null;
}
在高峰工作时间内，我每秒从 API 收到大约 400 - 500 条消息。这些消息以 XML 消息的形式出现。例如，一条消息可以包含多个订单，如下例所示。一个元素可以包含插入（创建）订单的操作，而另一个元素可以包含删除特定订单的操作。
<?xml version="1.0" encoding="utf-16"?>
<APIDATA xmlns="api-com">
<ORDER EngineID="0" PersistentOrderID="2791" ...>
<INSTSPECIFIER InstID="287" ... />
...
</ORDER>
<ORDER EngineID="0" PersistentOrderID="9840" ...>
<INSTSPECIFIER InstID="288" ... />
...
</ORDER>
RabbitMQ 服务器已配置，我使用 SSL 连接到它（带有证书和密钥）。 我使用 RabbitMQ.Client v6.2.1 连接到 RMQ。 交换、队列和绑定已经在 RabbitMQ 中定义。我的 Producer 应用程序只连接到它并开始发布。
我使用同步发布方法很重要，因为我们获取消息的顺序非常重要。例如，在一条消息中，我们得到一个创建订单的操作，而另一条消息紧随其后，告诉我们删除相同的订单。如果我使用 async 方法发布到 RMQ，我可能会在插入操作之前获得删除操作。
<?xml version="1.0" encoding="utf-16"?>
<APIDATA xmlns="api-com">
<ORDER ... PersistentOrderID="2791" OrderID="1234" ... Action="Insert" ...>
...
</ORDER>
删除消息：
<?xml version="1.0" encoding="utf-16"?>
<APIDATA xmlns="api-com">
<ORDER ... PersistentOrderID="2791" OrderID="1234" ... Action="Remove" ...>
...
</ORDER>
我使用以下方法发布到 RMQ：对象池（Microsoft 提供了一个名为 Microsoft.Extensions.ObjectPool 的包）-此处描述的方法-https://www.c-sharpcorner.com/article/publishing-rabbitmq-message-in-asp-net-core/。
我在这里使用以下代码：
class RabbitManager : IRabbitManager
{
private readonly DefaultObjectPool<IModel> _objectPool;
public RabbitManager(IPooledObjectPolicy<IModel> objectPolicy){
_objectPool = new DefaultObjectPool<IModel>(objectPolicy, Environment.ProcessorCount * 2);
}
public void Publish<T>(T message, string exchangeName, string exchangeType, string routeKey) where T : class {
if (message == null)
return;
var channel = _objectPool.Get();
try{
var sendBytes = Encoding.UTF8.GetBytes(message.ToString());
var properties = channel.CreateBasicProperties();
properties.ContentType = "application/json";
properties.DeliveryMode = 1; // Doesn't persist to disk
properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
channel.BasicPublish(exchangeName, routeKey, properties, sendBytes);
}
catch (Exception ex) {
throw ex;
}
finally {
_objectPool.Return(channel);
}
}
}
public class RabbitModelPooledObjectPolicy : IPooledObjectPolicy<IModel>
{
private readonly RabbitOptions _options;
private readonly IConnection _connection;
public RabbitModelPooledObjectPolicy(RabbitOptions _options){
this._options = _options;
_connection = GetConnection();
}
private IConnection GetConnection() {
var factory = new ConnectionFactory() {
HostName = _options.HostName,
UserName = _options.UserName,
Password = _options.Password,
//Port = _options.Port,
VirtualHost = _options.VHost,
};
if (!String.IsNullOrEmpty(_options.CertPath))
{
factory.RequestedConnectionTimeout = TimeSpan.FromMilliseconds(5000);
factory.Ssl.AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch | SslPolicyErrors.RemoteCertificateChainErrors;
factory.Ssl.CertificateValidationCallback += new RemoteCertificateValidationCallback(ValidateServerCertificate);
factory.Ssl.ServerName = _options.HostName;
factory.Ssl.CertPath = _options.CertPath;
factory.Ssl.CertPassphrase = _options.CertPass;
factory.Ssl.Version = SslProtocols.Tls12;
factory.Ssl.Enabled = true;
}
factory.RequestedHeartbeat = TimeSpan.FromSeconds(1);
factory.AutomaticRecoveryEnabled = true; // enable automatic connection recovery
factory.RequestedChannelMax = 32;
var _connection = factory.CreateConnection();
_connection.ConnectionShutdown += Connection_ConnectionShutdown;
return _connection;
}
private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs e){
Utils.log.Info("Connection broke!");
}
private bool ValidateServerCertificate(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors){
return true;
}
public IModel Create(){
return _connection.CreateModel();
}
public bool Return(IModel obj) {
if (obj.IsOpen) {
return true;
}
else {
obj?.Dispose();
return false;
}
}
}
下面是有问题的截图 - 不断增加的内存：
这是在上述屏幕截图之后拍摄的内存快照的堆栈跟踪：
在截取上面的屏幕截图后，我在控制台中的程序中收到以下错误消息：
26-04-2021 10:41:48 - OnNewData () RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure.
at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
at RabbitMQ.Client.Framing.Impl.Connection.EnsureIsOpen()
at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateModel()
at RabbitModelPooledObjectPolicy.Create() in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitModelPooledObjectPolicy.cs:line 77
at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Create()
at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Get()
at RabbitManager.Publish[T](T message, String exchangeName, String exchangeType, String routeKey) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitManager.cs:line 32
at ConsoleApp1.Service1.SendOrderToRMQ(JObject order) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 411
at ConsoleApp1.Service1.ParseXMLAnswer(String strOutputXML, String caller) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 372
at ConsoleApp1.Service1.OnNewData(String strXML) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 348
26-04-2021 10:41:48 - OnNewData () RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure.
at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
at RabbitMQ.Client.Framing.Impl.Connection.EnsureIsOpen()
at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateModel()
at RabbitModelPooledObjectPolicy.Create() in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitModelPooledObjectPolicy.cs:line 77
at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Create()
at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Get()
at RabbitManager.Publish[T](T message, String exchangeName, String exchangeType, String routeKey) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitManager.cs:line 32
at ConsoleApp1.Service1.SendOrderToRMQ(JObject order) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 411
at ConsoleApp1.Service1.ParseXMLAnswer(String strOutputXML, String caller) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 372
at ConsoleApp1.Service1.OnNewData(String strXML) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 348
这会导致程序最终崩溃：
我做了什么来防止内存泄漏：
- 这些类启用了 IDisposable 接口
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
// free managed resources
_onTimePerHour.Dispose();
_api.OnNewData -= OnNewData;
}
// free native resources if there are any.
}
- 在收到每条消息后强制进行垃圾回收：
private void ParseXMLAnswer(string strOutputXML, string caller) {
...
doc = null;
GC.Collect();
GC.WaitForPendingFinalizers();
}
这有点帮助，现在我的记忆问题在较长一段时间后会增加。
- 我使用 Visual Studio 的 ReShaper (enter link description here) 插件让我更好地了解内存问题的堆栈跟踪，但这并没有太大帮助。
我认为问题是什么：
RabbitMQ Producer 应用每秒获取许多消息，然后将其解析、拆分为多个 JSON 消息并使用同一通道发送到 RMQ。可能会出现以下情况：
- 我在单个 RMQ 通道上发布，不知何故我应该使用多个通道（一个连接但多个通道）
- 我收到的消息超出了我使用 RabbitMQ.Client .net 库通过 RMQ 解析和发送的数量
- 我在内存中为一些没有被释放的对象（可能是消息）保留了一些引用；
以前有人遇到过这个问题吗？因为我在任何地方都找不到有关此“SingleProducerSingleConsumerQueue+Segment
有人知道如何更深入地分析这个问题吗？
非常感谢！
编辑 1
我想需要更多信息来解决这个内存问题。
我有几个消费者使用来自 RabbitMQ 的数据（如 NodeJS 和 python 应用程序）。因此，我需要以通用方式设计 RabbitMQ 生产者，因为每个消费者需要不同的数据。每次我有一个新的消费者应用程序时，我都无法修改和重新启动我的 RabbitMQ 生产者。所以我需要以一种通用的方式发布我的消息。
例如，每个消费者都有自己的专用队列，并带有专用绑定。假设我有带有队列 cons1 和绑定的 consumer1：
- marketName.productName.*.1（productName 对应天数）。
此绑定是动态的，目前它对应于星期一（04.April），但明天它将对应于星期二（05.April）。
所以我需要通过使用将市场名称和产品名称存储在内存中
private static read-only Dictionary<string, List<string>> sequenceItemsHashed = new Dictionary<string, List<string>>();
private static readonly Dictionary<string, string> sequencesFromInstruments = new Dictionary<string, string>();
在我的逻辑中，我提到 sequenceItemsHashed 对应于 marketNames，sequenceFromInstruments 对应于 productNames。
这样我将所有消息发送到 RMQ，然后使用绑定在 RMQ 中进行排序。
编辑 2
据我了解，为了解决我的问题，我需要类似以下架构 (enter link description here)：
因此，我与 RMQ 服务器的单个连接中有多个线程，每个线程一个通道。
编辑 3
成功实现了管道、ConcurrentQueue 和推送到 RMQ 但仍然存在内存问题的消费者线程：
private readonly TransformBlock<string, string> orderFilter;
private readonly TransformBlock<string, JObject> xmlParser;
//private readonly TransformBlock<XmlDocument, JObject> xmlToJsonTransformer;
private readonly TransformManyBlock<JObject, JToken> jsonOrderFactory;
private readonly ActionBlock<JToken> messageSender;
ConcurrentQueue<JToken> concurrentQueue = new ConcurrentQueue<JToken>();
public Service1 (string [] args) {
...
// setup pipeline blocks
orderFilter = new TransformBlock<string, string>(FilterIncomingMessages);
xmlParser = new TransformBlock<string, JObject>(ParseXml);
jsonOrderFactory = new TransformManyBlock<JObject, JToken>(CreateOrderMessages);
messageSender = new ActionBlock<JToken>(SendMessage);
// build your pipeline
orderFilter.LinkTo(xmlParser, x => !string.IsNullOrEmpty(x));
orderFilter.LinkTo(DataflowBlock.NullTarget<string>()); // for non-order msgs
xmlParser.LinkTo(jsonOrderFactory);
jsonOrderFactory.LinkTo(messageSender, new DataflowLinkOptions { PropagateCompletion = true });
Task t2 = Task.Factory.StartNew(() =>
{
while (true) {
if (!concurrentQueue.IsEmpty)
{
JToken number;
while (concurrentQueue.TryDequeue(out number))
{
_rabbitMQ.PublishMessages(
Encoding.ASCII.GetBytes(number.ToString()),
"test"
);
}
} else
{
Thread.Sleep(1);
}
}
});
...
}
private string FilterIncomingMessages(string strXml){
if (strXml.Contains("<ORDER")) return strXml;
return null;
}
private JObject ParseXml(string strXml){
XmlDocument doc = new XmlDocument();
doc.LoadXml(strXml);
string jsonText = JsonConvert.SerializeXmlNode(doc);
var o = JObject.Parse(jsonText);
return o;
}
private IEnumerable<JToken> CreateOrderMessages(JObject o){
List<JToken> myList = new List<JToken>();
if (o.ContainsKey("GV8APIDATA")){
if (o["GV8APIDATA"]["ORDER"].Type is JTokenType.Object){
JToken order = o["GV8APIDATA"]["ORDER"];
myList.Add(order);
}
else if (o["GV8APIDATA"]["ORDER"].Type is JTokenType.Array){
JToken orders = o["GV8APIDATA"]["ORDER"];
foreach (var order in orders.Children()){
myList.Add(order);
}
}
}
return myList.ToArray ();
}
private void SendMessage(JToken order){
concurrentQueue.Enqueue(order);
}
新的解决方案有助于将逻辑分解为几个小部分，但我的内存仍在不断增加。
编辑 4
考虑到@Fildor 的回答，我做了以下事情：
我没有将包含带有
我删除了带有 Thread 和 ConcurrentQueue 的部分，并直接在最后一个 ActionBlock 中发布。
这解决了我的内存泄漏问题，但还有其他问题，例如：
- 如果消息足够大，我将只能打印大约 120 条消息/秒。如果我只打印简单的字符串“test”，我会得到 1780 条消息/秒的速率。
public Service1 (string [] args) {
...
// setup pipeline blocks
orderFilter = new TransformBlock<string, string>(FilterIncomingMessages);
xmlParser = new TransformBlock<string, OrdersResponse>(ParseXml);
jsonOrderFactory = new TransformManyBlock<OrdersResponse, Order>(CreateOrderMessages);
messageSender = new ActionBlock<Order>(SendMessage);
// build your pipeline
orderFilter.LinkTo(xmlParser, x => !string.IsNullOrEmpty(x));
orderFilter.LinkTo(DataflowBlock.NullTarget<string>()); // for non-order msgs
xmlParser.LinkTo(jsonOrderFactory);
jsonOrderFactory.LinkTo(messageSender, new DataflowLinkOptions { PropagateCompletion = true });
RunAsConsole(args);
}
private readonly TransformBlock<string, string> orderFilter;
private readonly TransformBlock<string, OrdersResponse> xmlParser;
private readonly TransformManyBlock<OrdersResponse, Order> jsonOrderFactory;
private readonly ActionBlock<Order> messageSender;
private void OnNewData(string strXML){
orderFilter.Post(strXML);
}
private string FilterIncomingMessages(string strXml){
if (strXml.Contains("<ORDER")) return strXml;
return null;
}
private OrdersResponse ParseXml(string strXml) {
var rootDataObj = DeserializeOrdersFromXML(strXml);
return rootDataObj;
}
private OrdersResponse DeserializeOrdersFromXML(string strOutputXML){
var xsExpirations = new XmlSerializer(typeof(OrdersResponse));
OrdersResponse rootDataObj = null;
using (TextReader reader = new StringReader(strOutputXML)) {
rootDataObj = (OrdersResponse)xsExpirations.Deserialize(reader);
reader.Close();
}
return rootDataObj;
}
private IEnumerable<Order> CreateOrderMessages(OrdersResponse o){
return o.orders;
}
private void SendMessage(Order order) {
_rabbitMQ.PublishMessages(
Encoding.ASCII.GetBytes(order.ToString()),
"test"
);
}
ORDER 对象看起来像：
[Serializable()]
[XmlRoot (ElementName = "ORDER")]
public class Order : IDisposable {
public void Dispose()
{
EngineID = null;
PersistentOrderID = null;
...
InstrumentSpecifier.Dispose();
InstrumentSpecifier = null;
GC.SuppressFinalize(this);
}
[XmlAttribute (AttributeName = "EngineID")]
public string EngineID { get; set; }
[XmlAttribute (AttributeName = "PersistentOrderID")]
public string PersistentOrderID { get; set; }
...
[XmlElement(ElementName = "INSTSPECIFIER")]
public InstrumentSpecifier InstrumentSpecifier { get; set; }
}
还有我的新 RabbitMQ 类：
public class RMQ : IDisposable {
private IConnection _connection;
public IModel Channel { get; private set; }
private readonly ConnectionFactory _connectionFactory;
private readonly string _exchangeName;
public RMQ (RabbitOptions _rabbitOptions){
try{
// _connectionFactory initialization
_connectionFactory = new ConnectionFactory()
{
HostName = _rabbitOptions.HostName,
UserName = _rabbitOptions.UserName,
Password = _rabbitOptions.Password,
VirtualHost = _rabbitOptions.VHost,
};
this._exchangeName = _rabbitOptions.ExchangeName;
if (!String.IsNullOrEmpty(_rabbitOptions.CertPath)){
_connectionFactory.RequestedConnectionTimeout = TimeSpan.FromMilliseconds(5000);
_connectionFactory.Ssl.AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch | SslPolicyErrors.RemoteCertificateChainErrors;
_connectionFactory.Ssl.CertificateValidationCallback += new RemoteCertificateValidationCallback(ValidateServerCertificate);
_connectionFactory.Ssl.ServerName = _rabbitOptions.HostName;
_connectionFactory.Ssl.CertPath = _rabbitOptions.CertPath;
_connectionFactory.Ssl.CertPassphrase = _rabbitOptions.CertPass;
_connectionFactory.Ssl.Version = SslProtocols.Tls12;
_connectionFactory.Ssl.Enabled = true;
}
_connectionFactory.RequestedHeartbeat = TimeSpan.FromSeconds(1);
_connectionFactory.AutomaticRecoveryEnabled = true; // enable automatic connection recovery
//_connectionFactory.RequestedChannelMax = 10;
if (_connection == null || _connection.IsOpen == false){
_connection = _connectionFactory.CreateConnection();
_connection.ConnectionShutdown += Connection_ConnectionShutdown;
}
if (Channel == null || Channel.IsOpen == false){
Channel = _connection.CreateModel();
}
Utils.log.Info("ConnectToRabbitMQ () Connecting to RabbitMQ. rabbitMQenvironment = ");
}
catch (Exception ex){
Utils.log.Error("Connection to RabbitMQ failed ! HostName = " + _rabbitOptions.HostName + " VirtualHost = " + _rabbitOptions.VHost);
Utils.printException("ConnectToRMQ ()", ex);
}
}
private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs e){
Utils.log.Info ("Connection broke!");
try{
if (ReconnectToRMQ()){
Utils.log.Info("Connected!");
}
}
catch (Exception ex){
Utils.log.Info("Connect failed!" + ex.Message);
}
}
private bool ReconnectToRMQ(){
if (_connection == null || _connection.IsOpen == false){
_connection = _connectionFactory.CreateConnection();
_connection.ConnectionShutdown += Connection_ConnectionShutdown;
}
if (Channel == null || Channel.IsOpen == false){
Channel = _connection.CreateModel();
return true;
}
return false;
}
private bool ValidateServerCertificate (object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors) {
return true;
}
public void DisconnectFromRMQ () {
Channel.Close ();
_connection.Close ();
}
public void Dispose(){
try{
Channel?.Close();
Channel?.Dispose();
Channel = null;
_connection?.Close();
_connection?.Dispose();
_connection = null;
}
catch (Exception e){
Utils.log.Error("Cannot dispose RabbitMQ channel or connection" + e.Message);
}
}
public void PublishMessages (byte [] message, string routingKey) {
if (this._connection == null || ! _connection.IsOpen) {
Utils.log.Error ("PublishMessages(), Connect failed! this.conn == null || !conn.IsOpen ");
ReconnectToRMQ();
} else {
var properties = Channel.CreateBasicProperties();
properties.Persistent = true;
Channel.BasicPublish (_exchangeName, routingKey, properties, message);
//serviceInstance1.Publish(message, _rabbitOptions.ExchangeName, "", routingKey);
}
}
}
现在有趣的是，如果我只将“test”之类的小字符串发布到 RabbitMQ 到我的预定义队列，我每秒可以发布超过 1780 条消息。
【问题讨论】：
-
“使用手动释放自己的内存：” - 这实际上不是你在做什么。您建议运行时执行 GC。如果你发现自己认为你需要这个，那么你有一个不同的问题。在这里，似乎是分配和/或引用时间过长。所以，我要做的是用足够的数据建立一个基准来获得真实的测量结果。然后调整所有已识别的瓶颈（一个接一个）并比较基准。不断改进，放弃没有任何改进的更改。
-
SetUpApi最多调用一次吗？如果不是，它被调用了多少次？如果你的记忆力如此有限，我也想知道你为什么要玩这么多字符串。您从字符串转到（旧）XmlDocument 只是为了将其序列化为 Json 字符串，以便您可以从中创建 JObject。我希望您只需转到 XDocument 一次，找到订单 XElement 节点，然后在该单个订单节点上调用 JsonConvert.SerializeXNode 即可减少字符串大小。除此之外，您可能仍需要按照 Fidor 的指示从事件线程中卸载您的工作。 -
“那么每次有一个来自公司的事件，它给了我api，就会调用ParseAnswer()方法”。在同一个线程上一个接一个。看看，这去哪儿了？
-
不，不要启动 1 个线程/消息。只需将事件处理与处理分离即可。
-
@rene 您是否尝试过一次进行一种操作来隔离泄漏到管道的某个阶段？此外，上面提到的管道架构可能会有所帮助（例如，使用带有专用消费者线程的内部并发队列进行转码步骤），另一个队列用于发送等。最后，您确定这是内存泄漏还是内存压力。这些是非常不同的。当应用恢复到稳定状态时会发生什么（一段时间没有消息？）...内存释放了还是保留了下来？
标签： c# .net memory-leaks rabbitmq