EasyNetQ 是一个容易使用,专门针对RabbitMQ的 .NET API。EasyNetQ是为了提供一个尽可能简洁的适用与RabbitMQ的.NET类库,下面看下怎么集成。
Q! T6 J) f9 U, g$ k" y3 U1、nuget 安装7 Q8 W0 \' q9 p/ P9 q
0 v; X6 o! T* l0 Q! `) U* e; s' l& a
2、配置连接串
: a+ Z* L/ C( |2 S3 c- i 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、这边我们构建一个消息体
4 I4 q& t5 K& t' Z% T) K" t) S/// <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、发送消息
7 S" z+ ^7 L( r /// <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、接收消息; k* Q" V/ E' R& n4 G
/// <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;
}
) ~0 l4 w. k9 m* o
在EasyNetQ中如果需要生产者确认功能,则需要在Rabbitmq的连接配置中设置publisherConfirms=true,这将会开启自动确认。在使用高级api定义交换机和队列时可以自己定义多种参数,比如消息是否持久化,消息最大长度等等。 |