RabbitMQ频道创建准则

| 我正在编写一个简单的类，我的应用程序将使用该类使用RabbitMQ发送和接收消息。我已经找到了尽可能多的关于RabbitMQ的方法，博客文章，白皮书和类似文章。大多数示例都显示使用包装在using块中的连接和通道，并且与之相反，您可能应该将它们实现为单例。特别是关于通道，我已经看到有评论说，同时使用单个通道的线程不应超过一个。我正在用C＃编写我的库。这是一个单例，在第一次实例化时连接了静态连接。我曾考虑过对频道进行相同的操作，但是我打算使用相同的库来允许发布/订阅多个交换/队列。发布和订阅都可以从多个线程完成。最后是我的问题：我应该如何实施渠道创建？每条消息？每个消费者都有一个独特的私人渠道，发布者将访问权同步到一个独特的渠道吗？你抓住我的漂移。请记住，我打算使用一台服务器，其中包含数十个使用者/发布者，但不多。谢谢！

已邀请:

5 个回复

翁茄口霉氖

编辑（2016-1-26）：通道不是线程安全的。在2015年4月至2015年5月之间，相关文档已更改。新文本：通道实例不得在线程之间共享。应用程序应该更喜欢每个线程使用一个Channel，而不是在多个线程之间共享同一Channel。尽管可以安全地并发调用通道上的某些操作，但某些操作却不能并发调用，这将导致错误的帧交错在电线上。线程之间共享通道也会干扰*发布者确认。从您的问题看来，您似乎没有预定义的固定数量的线程，这些线程大部分都可以发布/订阅RabbitMQ（在这种情况下，您可以考虑创建通道作为线程初始化的一部分，或者使用ThreadLocal<IModel> ）。如果很少有并发的RabbitMQ操作或消息大小总是很小，那么您可能只需在所有RabbitMQ发布/订阅操作周围放一个“ 1”就可以摆脱困境。如果您需要使用交错线程以交错方式传输多个请求-首先是通道的用途-使用任意线程，则可能需要创建一个通道池，例如ConcurrentQueue<IModel>，在其中排队未使用的频道，并在需要的时间出队。通道创建的开销很低，从性能测试来看，我觉得通道创建过程不涉及任何网络io，即，似乎在客户端首次使用RabbitMQ服务器时会自动创建通道。 OLD（2016年1月26日之前）：Java和.net实现的现在大部分已过时的详细信息： Re：通道和多个线程，由于它依赖于实现，因此有点混乱。 Java实现：通道是线程安全的：通道实例可安全用于多个线程。但：在多个线程之间共享通道时，确认操作不正确 .net实现：通道不是线程安全的：如果一个以上的线程需要访问特定的IModel实例，则应用程序应自己强制执行互斥。 IModel操作序列化不正确的症状包括但不限于： •电线上发送的帧序列无效 •抛出NotSupportedExceptions ... 因此，除了.bin有用的答案（无论它是否线程安全）都适用外，在.net实现中，您不能只共享一个通道。

吐兄

它阐明了aqmp内部。目前，我的理解是：答：我可以保持从每个应用程序到服务器的单个共享tcp连接（作为静态共享资源） B.我应该为每个“任务”创建一个通道（一个用于侦听队列X，一个用于发布以交换Y，等等，假设这些“任务”可以并行执行） C.或者我可以在单个应用程序中为所有通道使用一个通道，同时确保对它的访问是同步的-使用某种锁定机制，假设使用（锁定）通道的实际时间相对较短。谢谢。

炬卤遁蝎变

我无法评论C＃实现的细节，但可能会有助于了解Amqp通道旨在共享单个TCP连接，即启用多路复用。单个通道只能一次发送一个或一次接收一条消息，但是连接可以同时在不同的通道上接收消息。如果您有2个通过Amqp发送给单个使用者的1GB大文件，则可能会将邮件分成1万个大块并以交错方式发送。设置连接时，您可以控制默认的Amqp消息大小，这取决于是否和何时发生交错。 AFAIK此功能旨在在多个使用者共享一个连接并且一个使用者收到大消息时帮助防止饥饿。 HTH。

募磷

您是正确的，该Channel不是线程安全的，不应被一个以上的线程访问。如果您计划使用不同的队列，则可以在一个通道中使用多个队列。但是，如果您计划进行多个交换（不确定为什么要在这里增加多个交换。），您将不得不跟踪单身人士中的多个交换和渠道，或者将这一责任转移给呼叫者。我将建立您的具有观察者模式的单例，并在其中保存所有RabbitMQ内容以及对您的调用者的引用。然后，您还需要编组处理此问题的单个线程的调用，否则可能会冒着通道对象问题的风险。

慰泥悍瓶

使用ASP.NET Core，您可以利用ObjectPool。创建一个IPooledObjectPolicy

    using Microsoft.Extensions.ObjectPool;  
    using Microsoft.Extensions.Options;  
    using RabbitMQ.Client;  

    public class RabbitModelPooledObjectPolicy : IPooledObjectPolicy<IModel>  
    {  
        private readonly RabbitOptions _options;  

        private readonly IConnection _connection;  

        public RabbitModelPooledObjectPolicy(IOptions<RabbitOptions> optionsAccs)  
        {  
            _options = optionsAccs.Value;  
            _connection = GetConnection();  
        }  

        private IConnection GetConnection()  
        {  
            var factory = new ConnectionFactory()  
            {  
                HostName = _options.HostName,  
                UserName = _options.UserName,  
                Password = _options.Password,  
                Port = _options.Port,  
                VirtualHost = _options.VHost,  
            };  

            return factory.CreateConnection();  
        }  

        public IModel Create()  
        {  
            return _connection.CreateModel();  
        }  

        public bool Return(IModel obj)  
        {  
            if (obj.IsOpen)  
            {  
                return true;  
            }  
            else  
            {  
                obj?.Dispose();  
                return false;  
            }  
        }  
    }

然后为ObjectPool配置依赖项注入

services.AddSingleton<ObjectPoolProvider, DefaultObjectPoolProvider>();
services.AddSingleton(s =>
{
   var provider = s.GetRequiredService<ObjectPoolProvider>();
   return provider.Create(new RabbitModelPooledObjectPolicy());
});

然后可以注入in5ѭ并使用它

var channel = pool.Get();
try
{
    channel.BasicPublish(...);
}
finally
{
    pool.Return(channel);
}

资料来源： https://www.c-sharpcorner.com/article/publishing-rabbitmq-message-in-asp-net-core/ https://developpaper.com/detailed-explanation-of-object-pools-various-usages-in-net-core/

要回复问题请先登录或注册

RabbitMQ频道创建准则

5 个回复

发起人

rabbitmq

c#

messaging

问题状态

RabbitMQ频道创建准则

与内容相关的链接

5 个回复

发起人

rabbitmq

c#

messaging

问题状态