一起谈.NET技术，NET 下RabbitMQ实践 [实战篇]

/// <summary>
/// IHttpModuleErrLogService 接口类
/// </summary>
[ServiceContract]
public interface IHttpModuleErrLogService
{
/// <summary>
/// 添加 httpModuleErrLogData日志信息
/// </summary>
/// <param name="httpModuleErrLogData"></param>
[OperationContract]
void AddLog(HttpModuleErrLogData httpModuleErrLogData);
}

[ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)] // Single - 为所有客户端调用分配一个服务实例。
public class HttpModuleErrLogService : IHttpModuleErrLogService
{
/// <summary>
/// 获取 HttpModuleErrLogInfo配置文件对象实例
/// </summary>
private static HttpModuleErrLogInfo httpModuleErrorLogInfo = RabbitMQConfigs.GetConfig().HttpModuleErrLog;
/// <summary>
/// 定时器对象
/// </summary>
private static System.Timers.Timer _timer;
/// <summary>
/// 定时器的时间
/// </summary>
private static int _elapsed = 0 ;


public static void Initial(System.Windows.Forms.RichTextBox msgBox, int elapsed)
{
_msgBox = msgBox;
_elapsed = elapsed;


// 初始定时器
if (_elapsed > 0 )
{
_timer = new System.Timers.Timer() { Interval = elapsed * 1000 , Enabled = true , AutoReset = true };
_timer.Elapsed += new System.Timers.ElapsedEventHandler(Timer_Elapsed);
_timer.Start();
}
}


/// <summary>
/// 时间到时执行出队操作
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private static void Timer_Elapsed( object sender, System.Timers.ElapsedEventArgs e)
{
Dequeue();
}

/// <summary>
/// 添加 httpModuleErrLogData日志信息
/// </summary>
/// <param name="httpModuleErrLogData"></param>
public void AddLog(HttpModuleErrLogData httpModuleErrLogData)
{
Enqueue(httpModuleErrLogData);


if (_elapsed <= 0 ) // 如果使用定时器（为0 时）,则立即执行出队操作
Dequeue();
}


/// <summary>
/// 交换机名称
/// </summary>
private const string EXCHANGE = " ex1 " ;
/// <summary>
/// 交换方法,更多内容参见: http://melin.javaeye.com/blog/691265
/// </summary>
private const string EXCHANGE_TYPE = " direct " ;
/// <summary>
/// 路由key,更多内容参见: http://sunjun041640.blog.163.com/blog/static/256268322010328102029919/
/// </summary>
private const string ROUTING_KEY = " m1 " ;


/// <summary>
/// 日志入队
/// </summary>
/// <param name="httpModuleErrLogData"></param>
public static void Enqueue(HttpModuleErrLogData httpModuleErrLogData)
{
Uri uri = new Uri(httpModuleErrorLogInfo.RabbitMQAddress);
ConnectionFactory cf = new ConnectionFactory()
{
UserName = httpModuleErrorLogInfo.UserName,
Password = httpModuleErrorLogInfo.PassWord,
VirtualHost = " dnt_mq " ,
RequestedHeartbeat = 0 ,
Endpoint = new AmqpTcpEndpoint(uri)
};
using (IConnection conn = cf.CreateConnection())
{
using (IModel ch = conn.CreateModel())
{
if (EXCHANGE_TYPE != null )
{
ch.ExchangeDeclare(EXCHANGE, EXCHANGE_TYPE); // ,true,true,false,false, true,null);
ch.QueueDeclare(httpModuleErrorLogInfo.QueueName, true ); // true, true, true, false, false, null);
ch.QueueBind(httpModuleErrorLogInfo.QueueName, EXCHANGE, ROUTING_KEY, false , null );
}
IMapMessageBuilder b = new MapMessageBuilder(ch);
IDictionary target = b.Headers;
target[ " header " ] = " HttpErrLog " ;
IDictionary targetBody = b.Body;
targetBody[ " body " ] = SerializationHelper.Serialize(httpModuleErrLogData);
((IBasicProperties)b.GetContentHeader()).DeliveryMode = 2 ; // persistMode
ch.BasicPublish(EXCHANGE, ROUTING_KEY,
(IBasicProperties)b.GetContentHeader(),
b.GetContentBody());
}
}
}

/// <summary>
/// 日志出队
/// </summary>
public static void Dequeue()
{
string serverAddress = httpModuleErrorLogInfo.RabbitMQAddress.Replace( " amqp:// " , "" ).TrimEnd( ' / ' ); // "10.0.4.85:5672";
ConnectionFactory cf = new ConnectionFactory()
{
UserName = httpModuleErrorLogInfo.UserName,
Password = httpModuleErrorLogInfo.PassWord,
VirtualHost = " dnt_mq " ,
RequestedHeartbeat = 0 ,
Address = serverAddress
};

using (IConnection conn = cf.CreateConnection())
{
using (IModel ch = conn.CreateModel())
{
while ( true )
{
BasicGetResult res = ch.BasicGet(httpModuleErrorLogInfo.QueueName, false );
if (res != null )
{
try
{
string objstr = System.Text.UTF8Encoding.UTF8.GetString(res.Body).Replace( " \0\0\0