C# Windows Azure Queue的操作

<appSettings>
<add key="StorageConnectionString" value="your connection string" />
</appSettings>

namespace Axe.AzureStorage
{
using System;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading;
using System.Threading.Tasks;

using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Queue;

public class WinAzureStorageAsync
{
private readonly CloudQueue queue;
private readonly int timeoutSecond;
private CloudQueueClient queueClient;
public CloudQueueClient QueueClient
{
get
{
if (this.queueClient != null)
return this.queueClient;

var storageAccount = CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("StorageConnectionString"));
this.queueClient = storageAccount.CreateCloudQueueClient();
return this.queueClient;
}
}

////since each time fetch message is not a block operation
////so need to set a timeout & keep fetching , default is 3 seconds
private const int SleepInterval = 100;

public WinAzureStorageAsync(string queueName, int timeoutSecond = 3)
{
queueName = queueName.ToLower();
this.queue = this.QueueClient.GetQueueReference(queueName);
if (!this.QueueClient.GetQueueReference(queueName).Exists())
{
this.queue.CreateIfNotExists();
}

this.timeoutSecond = timeoutSecond;
}

public async Task<CloudQueueMessage> GetMessage()
{
CloudQueueMessage message = null;

var passed = 0;

while (message == null && passed < this.timeoutSecond * 10 * SleepInterval)
{
message = await this.queue.GetMessageAsync();
Thread.Sleep(SleepInterval);
passed += SleepInterval;
}
if (message == null)
{
throw new TimeoutException("Get Message From Azure Queue Operation has been timeout");
}
await this.queue.DeleteMessageAsync(message);

return message;
}

public async Task<string> GetString()
{
var msg = await this.GetMessage();
return msg.AsString;
}

public async Task<byte[]> GetBytes()
{
var msg = await this.GetMessage();
return msg.AsBytes;
}

public T Get<T>() where T : new()
{
var bytes = this.GetBytes();
return this.BytesToT<T>(bytes.Result);
}

public async Task Add(string message)
{
await this.queue.AddMessageAsync(new CloudQueueMessage(message));
}

public async Task Add(byte[] bytes)
{
await this.queue.AddMessageAsync(new CloudQueueMessage(bytes));
}

public void Add<T>(T obj) where T : new()
{
var bytes = this.TToBytes(obj);
this.Add(bytes);
}

/// <summary>
/// Note : this operation make takes around 40 seconds to complete, reference here:
/// http://msdn.microsoft.com/library/azure/dd179387.aspx /// </summary>
/// <returns></returns>
public async Task DeleteIfExists()
{
await this.queue.DeleteIfExistsAsync();
}

public async Task<bool> IsExist(string queueName)
{
queueName = queueName.ToLower();
return await this.QueueClient.GetQueueReference(queueName).ExistsAsync();
}

public void ClearMessage()
{
this.queue.Clear();
}

private T BytesToT<T>(byte[] bytes)
{
using (var ms = new MemoryStream())
{
ms.Write(bytes, 0, bytes.Length);
var bf = new BinaryFormatter();
ms.Position = 0;
var x = bf.Deserialize(ms);
return (T)x;
}
}

private byte[] TToBytes<T>(T obj)
{
var bf = new BinaryFormatter();
using (var ms = new MemoryStream())
{
bf.Serialize(ms, obj);
return ms.ToArray();
}
}
}
}