消息队列、OSS常用操作封装

public class MessageQueue
{
#region Private Properties
private const string _accessKeyId = "";
private const string _secretAccessKey = "";
private const string _endpoint = "";

private static string _queueName;
private const int _receiveTimes = 1;
private const int _receiveInterval = 2;
private const int batchSize = 6;

#endregion

#region 设置队列名称
public string queueName
{
set
{
_queueName = value;

}
get
{
return _queueName;
}
}
#endregion

#region 判断消息队列是否存在
/// <summary>
/// 判断消息队列是否存在
/// </summary>
/// <returns></returns>
public static bool QueueIsExist()
{
bool flag = false;
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var nativeQueue = client.GetNativeQueue(_queueName);
var getQueueAttributesResponse = nativeQueue.GetAttributes();
flag = true;
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
return flag;
}
#endregion

#region 创建消息队列
/// <summary>
/// 创建消息队列
/// </summary>
/// <returns></returns>
public static string CreateQueue()
{
/*
DelaySeconds	发送到该 Queue 的所有消息默认将以DelaySeconds参数指定的秒数延后可被消费，单位为秒。	0-604800秒（7天）范围内某个整数值，默认值为0
MaximumMessageSize	发送到该Queue的消息体的最大长度，单位为byte。	1024(1KB)-65536（64KB）范围内的某个整数值，默认值为65536（64KB）。
MessageRetentionPeriod	消息在该 Queue 中最长的存活时间，从发送到该队列开始经过此参数指定的时间后，不论消息是否被取出过都将被删除，单位为秒。	60 (1分钟)-1296000 (15 天)范围内某个整数值，默认值345600 (4 天)
VisibilityTimeout	消息从该 Queue 中取出后从Active状态变成Inactive状态后的持续时间，单位为秒。	1-43200(12小时)范围内的某个值整数值，默认为30（秒）
PollingWaitSeconds	当 Queue 中没有消息时，针对该 Queue 的 ReceiveMessage 请求最长的等待时间，单位为秒。	0-30秒范围内的某个整数值，默认为0（秒）
LoggingEnabled	是否开启日志管理功能，True表示启用，False表示停用	True/False，默认为False
*/
string queueName = string.Empty;
var createQueueRequest = new CreateQueueRequest
{
QueueName = _queueName,
Attributes =
{
DelaySeconds = 0,
MaximumMessageSize = 65536,
MessageRetentionPeriod = 345600,
VisibilityTimeout = 3600,
PollingWaitSeconds = 3
}
};
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var queue = client.CreateQueue(createQueueRequest);
queueName = queue.QueueName;
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
Thread.Sleep(2000);
return queueName;
}
#endregion

#region 删除消息队列
/// <summary>
/// 删除消息队列
/// </summary>
/// <returns></returns>
public static bool DeleteQueue()
{
bool flag = false;
var deleteQueueRequest = new DeleteQueueRequest(_queueName);
deleteQueueRequest.AddHeader("Accept", "IE6"); //Add extra request headers
//deleteQueueRequest.AddParameter("param1", "value1"); //InvalidQueryString
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var deleteQueueResponse = client.DeleteQueue(deleteQueueRequest);
flag = true;
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
return flag;
}
#endregion

#region 发送消息(单条或多条)
/// <summary>
/// 发送消息(单条或多条)
/// </summary>
/// <param name="models">SendMessageRequest集合</param>
/// <returns></returns>
public static bool BathSendMessage(List<SendMessageRequest> models)
{
bool flag = false;
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var nativeQueue = client.GetNativeQueue(_queueName);
List<SendMessageRequest> requests = new List<SendMessageRequest>();
for (int i = 0; i < models.Count; i++)
{
requests.Add(models[i]);
}
BatchSendMessageRequest batchSendRequest = new BatchSendMessageRequest()
{
Requests = requests
};
var sendMessageResponse = nativeQueue.BatchSendMessage(batchSendRequest);
flag = true;
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
return flag;
}
#endregion

#region 消费消息(单条或多条)
/// <summary>
/// 消费消息(单条或多条)
/// </summary>
/// <param name="itemNum">数目</param>
/// <returns></returns>
public static List<Message> ReceiveMessage(int itemNum)
{
List<Message> lists = new List<Message>();
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var nativeQueue = client.GetNativeQueue(_queueName);
for (int i = 0; i < itemNum; i++)
{
var receiveMessageResponse = nativeQueue.ReceiveMessage();
Message message = receiveMessageResponse.Message;
lists.Add(message);
}
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
return lists;
}
#endregion

#region 删除消息
/// <summary>
/// 删除消息
/// </summary>
/// <param name="receiptHandle">receiptHandle</param>
/// <returns></returns>
public static bool DeleteMessage(string receiptHandle)
{
bool flag = false;
var deletedReceiptHandle = receiptHandle;
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var nativeQueue = client.GetNativeQueue(_queueName);
var deleteMessageResponse = nativeQueue.DeleteMessage(receiptHandle);
flag = true;
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
return flag;
}
#endregion

#region 修改消息可见时间
/// <summary>
/// 修改消息可见时间
/// </summary>
/// <param name="receiptHandle">receiptHandle</param>
/// <param name="visibilityTimeout">从现在到下次可被用来消费的时间间隔</param>
/// <returns></returns>
public static bool ChangeMessageVisibility(string receiptHandle, int visibilityTimeout)
{
bool flag = false;
var deletedReceiptHandle = receiptHandle;
try
{
using (IMNS client = new Aliyun.MNS.MNSClient(_accessKeyId, _secretAccessKey, _endpoint))
{
var nativeQueue = client.GetNativeQueue(_queueName);
var changeMessageVisibilityRequest = new ChangeMessageVisibilityRequest
{
ReceiptHandle = receiptHandle,
VisibilityTimeout = visibilityTimeout
};
var changeMessageVisibilityResponse = nativeQueue.ChangeMessageVisibility(changeMessageVisibilityRequest);
flag = true;
}
}
catch (Exception ex) { Console.WriteLine(ex.ToString()); }
return flag;
}
#endregion
}

using System;
using System.IO;
using System.Text;
using System.Threading;
using System.Security.Cryptography;
using Aliyun.OSS.Common;

namespace Aliyun.OSS.Samples
{
/// <summary>
/// 获取OSS对象
/// </summary>
public static class GetObjectSample
{
static string accessKeyId = Config.AccessKeyId;
static string accessKeySecret = Config.AccessKeySecret;
static string endpoint = Config.Endpoint;
static OssClient client = new OssClient(endpoint, accessKeyId, accessKeySecret);

static string key = "123456.jpg";
static string fileToUpload = Config.FileToUpload;
static string dirToDownload = Config.DirToDownload;

static AutoResetEvent _event = new AutoResetEvent(false);

public static void GetObjects(string bucketName)
{
GetObject(bucketName); //获取文件

GetObjectByRequest(bucketName);

AsyncGetObject(bucketName); //异步方式获取文件
}

public static void GetObject(string bucketName)
{
try
{
client.PutObject(bucketName, key, fileToUpload);

var result = client.GetObject(bucketName, key);

using (var requestStream = result.Content)
{
using (var fs = File.Open(Path.Combine(dirToDownload, key), FileMode.OpenOrCreate))
{
int length = 4 * 1024;
var buf = new byte[length];
do
{
length = requestStream.Read(buf, 0, length);
fs.Write(buf, 0, length);
} while (length != 0);
}
}

Console.WriteLine("Get object succeeded");
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void GetObjectByRequest(string bucketName)
{
try
{
client.PutObject(bucketName, key, fileToUpload);

var request = new GetObjectRequest(bucketName, key);
request.SetRange(0, 100);

var result = client.GetObject(request);

Console.WriteLine("Get object succeeded, length:{0}", result.Metadata.ContentLength);
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void AsyncGetObject(string bucketName)
{
const string key = "AsyncGetObject";
try
{
client.PutObject(bucketName, key, fileToUpload);

string result = "Notice user: put object finish";
client.BeginGetObject(bucketName, key, GetObjectCallback, result.Clone());

_event.WaitOne();
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

private static void GetObjectCallback(IAsyncResult ar)
{
try
{
var result = client.EndGetObject(ar);

using (var requestStream = result.Content)
{
using (var fs = File.Open(dirToDownload + "/sample2.data", FileMode.OpenOrCreate))
{
int length = 4 * 1024;
var buf = new byte[length];
do
{
length = requestStream.Read(buf, 0, length);
fs.Write(buf, 0, length);
} while (length != 0);
}
}

Console.WriteLine(ar.AsyncState as string);
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
finally
{
_event.Set();
}
}
}
}

using System;
using System.Collections.Generic;
using Aliyun.OSS.Common;

namespace Aliyun.OSS.Samples
{
/// <summary>
/// 删除OSS对象
/// </summary>
public static class DeleteObjectsSample
{
static string accessKeyId = Config.AccessKeyId;
static string accessKeySecret = Config.AccessKeySecret;
static string endpoint = Config.Endpoint;
static OssClient client = new OssClient(endpoint, accessKeyId, accessKeySecret);

public static void DeleteObject(string bucketName)
{
try
{
string key = null;
var listResult = client.ListObjects(bucketName);
foreach (var summary in listResult.ObjectSummaries)
{
key = summary.Key;
break;
}

client.DeleteObject(bucketName, key);

Console.WriteLine("Delete object succeeded");
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void DeleteObjects(string bucketName)
{
try
{
var keys = new List<string>();
var listResult = client.ListObjects(bucketName);
foreach (var summary in listResult.ObjectSummaries)
{
keys.Add(summary.Key);
break; //不跳出删除全部
}
var request = new DeleteObjectsRequest(bucketName, keys, false);
client.DeleteObjects(request);

Console.WriteLine("Delete objects succeeded");
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}
}
}

using System;
using System.IO;
using System.Threading;
using Aliyun.OSS.Common;
using System.Text;
using Aliyun.OSS.Util;

namespace Aliyun.OSS.Samples
{
/// <summary>
/// 上传文件或对象到OSS
/// </summary>
public static class PutObjectSample
{
static string accessKeyId = Config.AccessKeyId;
static string accessKeySecret = Config.AccessKeySecret;
static string endpoint = Config.Endpoint;
static OssClient client = new OssClient(endpoint, accessKeyId, accessKeySecret);

static string fileToUpload = Config.FileToUpload;

static AutoResetEvent _event = new AutoResetEvent(false);

/// <summary>
/// sample for put object to oss
/// </summary>
public static void PutObject(string bucketName)
{
PutObjectFromFile(bucketName); //上传文件

PutObjectFromString(bucketName); //上传String

PutObjectWithDir(bucketName); //创建目录上传

PutObjectWithMd5(bucketName); //MD5验证上传

PutObjectWithHeader(bucketName); //设置Header上传

AsyncPutObject(bucketName); //异步上传
}

public static void PutObjectFromFile(string bucketName)
{
const string key = "PutObjectFromFile";
try
{
client.PutObject(bucketName, key, fileToUpload);
Console.WriteLine("Put object:{0} succeeded", key);
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void PutObjectFromString(string bucketName)
{
const string key = "PutObjectFromString";
const string str = "Aliyun OSS SDK for C#";

try
{
byte[] binaryData = Encoding.ASCII.GetBytes(str);
var stream = new MemoryStream(binaryData);

client.PutObject(bucketName, key, stream);
Console.WriteLine("Put object:{0} succeeded", key);
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void PutObjectWithDir(string bucketName)
{
const string key = "folder/sub_folder/PutObjectFromFile";

try
{
client.PutObject(bucketName, key, fileToUpload);
Console.WriteLine("Put object:{0} succeeded", key);
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void PutObjectWithMd5(string bucketName)
{
const string key = "PutObjectWithMd5";

string md5;
using (var fs = File.Open(fileToUpload, FileMode.Open))
{
md5 = OssUtils.ComputeContentMd5(fs, fs.Length);
}

var meta = new ObjectMetadata() { ContentMd5 = md5 };
try
{
client.PutObject(bucketName, key, fileToUpload, meta);

Console.WriteLine("Put object:{0} succeeded", key);
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void PutObjectWithHeader(string bucketName)
{
const string key = "PutObjectWithHeader";
try
{
using (var content = File.Open(fileToUpload, FileMode.Open))
{
var metadata = new ObjectMetadata();
metadata.ContentLength = content.Length;

metadata.UserMetadata.Add("github-account", "qiyuewuyi");

client.PutObject(bucketName, key, content, metadata);

Console.WriteLine("Put object:{0} succeeded", key);
}
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

public static void AsyncPutObject(string bucketName)
{
const string key = "AsyncPutObject";
try
{
// 1. put object to specified output stream
using (var fs = File.Open(fileToUpload, FileMode.Open))
{
var metadata = new ObjectMetadata();
metadata.UserMetadata.Add("mykey1", "myval1");
metadata.UserMetadata.Add("mykey2", "myval2");
metadata.CacheControl = "No-Cache";
metadata.ContentType = "text/html";

string result = "Notice user: put object finish";
client.BeginPutObject(bucketName, key, fs, metadata, PutObjectCallback, result.ToCharArray());

_event.WaitOne();
}
}
catch (OssException ex)
{
Console.WriteLine("Failed with error code: {0}; Error info: {1}. \nRequestID:{2}\tHostID:{3}",
ex.ErrorCode, ex.Message, ex.RequestId, ex.HostId);
}
catch (Exception ex)
{
Console.WriteLine("Failed with error info: {0}", ex.Message);
}
}

private static void PutObjectCallback(IAsyncResult ar)
{
try
{
client.EndPutObject(ar);

Console.WriteLine(ar.AsyncState as string);
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
finally
{
_event.Set();
}
}
}
}