EasyNetQ操作RabbitMQ

Posted 2021-12-31 dotNET跨平台

EasyNetQ 是一个容易使用，专门针对RabbitMQ的 .NET API。EasyNetQ是为了提供一个尽可能简洁的适用与RabbitMQ的.NET类库。

下面看下怎么集成。

1、nuget 安装

2、配置连接串

public static IBus CreateMessageBus()
        
            // 消息服务器连接字符串
            var connectionString = ConfigurationManager.ConnectionStrings["RabbitMQConnString"];
                 
            if (connectionString == null || connectionString.ConnectionString == string.Empty)
            
                throw new Exception("messageserver connection string is missing or empty");
            
            return RabbitHutch.CreateBus(connectionString.ConnectionString);
        

3、这边我们构建一个消息体

/// <summary>
    /// 消息类实体
    /// </summary>
    [Serializable]
    public class RabbitMQ_Message
    
        public RabbitMQ_Message()
        
            MessageID = DateTime.Now.Ticks.ToString();
        
        /// <summary>
        /// 消息id
        /// </summary>
        public string MessageID  get; set; 
        /// <summary>
        /// 消息标题
        /// </summary>
        public string MessageTitle  get; set; 
        /// <summary>
        /// 消息内容
        /// </summary>
        public string MessageBody  get; set; 
        /// <summary>
        /// 消息管道
        /// </summary>
        public RabbitMessageRouterEnum MessageRouter  get; set; 

        /// <summary>
        /// 游客id
        /// </summary>
        public int customerId  get; set; 
        /// <summary>
        /// 标示代码 0:正确
        /// </summary>
        public ResponseStatus result  get; set; 

        /// <summary>
        /// 消息类型
        /// </summary>
        public SuperSocketMessageTypeEnum superSocketMessageType  get; set; 

        /// <summary>
        /// 消息过期时间（毫秒）
        /// </summary>
        public int expiredMillSeconds  get; set; 
    

4、发送消息

/// <summary>
        /// 发送消息
        /// </summary>
        public static void Publish(RabbitMQ_Message msg)
        
             创建消息bus
            IBus bus = null;
            try
            
                 创建消息bus
                bus = BusBuilder.CreateMessageBus();
                bus.Publish(msg, x =>
                
                    x.WithTopic($"msg.MessageRouter.ToDescription().msg.customerId");
                    if (msg.expiredMillSeconds > 0)
                    
                        x.WithExpires(msg.expiredMillSeconds);
                    
                );//通过管道发送消息               
                LogExtention.getInstance().WriteCustomLogAsync(msg, "RabbitMQ消息发送", "MQHelperPublish");
            
            catch (EasyNetQException ex)
            
                LogExtention.getInstance().ErrorAsync(ex, "RabbitMQ--MQHelper--Publish发布消息时出错");
                //处理连接消息服务器异常 
            
            finally
            
                if (bus != null)
                
                    bus.Dispose();//与数据库connection类似，使用后记得销毁bus对象
                               
            
        

4、接收消息

/// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="msg"></param>
        public static ISubscriptionResult Subscribe(RabbitMQ_Message msg, IProcessMessage ipro)
        
             创建消息bus
            IBus bus = null;
            try
            
                bus = BusBuilder.CreateMessageBus();
                //subscriptionId设置不同的话，每一个subscriptionId都会收到相同的消息，下面的写法只会有一个接收者
                var subscriptionResult = bus.Subscribe<RabbitMQ_Message>(msg.MessageRouter.ToDescription(), message => ipro.ProcessMsg(message),
                    x => x.WithQueueName(msg.customerId.ToString()).WithTopic($"msg.MessageRouter.ToDescription().msg.customerId"));
                //subscriptionResult.Dispose();//取消订阅
                return subscriptionResult;
            
            catch (EasyNetQException ex)
            
                LogExtention.getInstance().ErrorAsync(ex, "RabbitMQ--MQHelper--Subscribe订阅消息时出错");
                //处理连接消息服务器异常 
            
            finally
            
                if (bus != null)
                
                    bus.Dispose();//与数据库connection类似，使用后记得销毁bus对象
                
            
            return null;
        

在EasyNetQ中如果需要生产者确认功能，则需要在Rabbitmq的连接配置中设置publisherConfirms=true，这将会开启自动确认。在使用高级api定义交换机和队列时可以自己定义多种参数，比如消息是否持久化，消息最大长度等等。