EasyNetQ 是一个容易使用,专门针对RabbitMQ的 .NET API。EasyNetQ是为了提供一个尽可能简洁的适用与RabbitMQ的.NET类库,下面看下怎么集成。
+ i1 _& _4 k0 q c6 W1、nuget 安装. Z" z4 K; Y7 L5 p, c
# P/ c) N1 i$ h8 a; I6 d
2、配置连接串
5 _! I9 f4 ] K# c0 F
- 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、这边我们构建一个消息体. E' o/ I6 T( i
- /// <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、发送消息9 f; d: C2 [$ i5 ^( r" y: P& |2 q
- /// <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、接收消息) l( O1 H: l/ A
- /// <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;
- }
$ D0 H! f& K- {5 W7 ~
在EasyNetQ中如果需要生产者确认功能,则需要在Rabbitmq的连接配置中设置publisherConfirms=true,这将会开启自动确认。在使用高级api定义交换机和队列时可以自己定义多种参数,比如消息是否持久化,消息最大长度等等。 |