QQ登录

只需要一步,快速开始

APP扫码登录

只需要一步,快速开始

手机号码,快捷登录

手机号码,快捷登录

查看: 3498|回复: 0

[C#/.NET] EasyNetQ操作RabbitMQ

[复制链接]

等级头衔

积分成就    金币 : 2851
   泡泡 : 1516
   精华 : 6
   在线时间 : 1305 小时
   最后登录 : 2024-12-7

丰功伟绩

优秀达人突出贡献荣誉管理论坛元老活跃会员

联系方式
发表于 2021-12-7 11:52:51 | 显示全部楼层 |阅读模式
       EasyNetQ 是一个容易使用,专门针对RabbitMQ的 .NET API。EasyNetQ是为了提供一个尽可能简洁的适用与RabbitMQ的.NET类库,下面看下怎么集成。
: w' Q# o9 N; {$ \* A1、nuget 安装3 S# R, ^4 @# A" W6 A6 R8 w4 [7 I" w
1.jpg
& _% Z& L. ~3 Z3 k2、配置连接串
  Q& n4 t. q" l5 E% k
   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、这边我们构建一个消息体( v! b+ Z& L. w3 c& P+ C& d6 j
/// <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、发送消息
, K- R* ~3 T! r8 S
  /// <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、接收消息
1 X8 b( t( g: s' r$ @! c
    /// <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;
        }
2.jpg
1 H# D/ [8 O0 \) t0 B       在EasyNetQ中如果需要生产者确认功能,则需要在Rabbitmq的连接配置中设置publisherConfirms=true,这将会开启自动确认。在使用高级api定义交换机和队列时可以自己定义多种参数,比如消息是否持久化,消息最大长度等等。
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

QQ|手机版|小黑屋|paopaomj.COM ( 渝ICP备18007172号|渝公网安备50010502503914号 )

GMT+8, 2024-12-7 03:14

Powered by paopaomj X3.5 © 2016-2025 sitemap

快速回复 返回顶部 返回列表