阿里云消息队列MQ_HTTP接入 for .NetCore 简单例子

简介:
复制代码
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Security.Cryptography;
using System.Text;
using System.Threading.Tasks;
using Newtonsoft.Json;

namespace MQWebCore
{
public class MQHelper
{
string URL = " http://publictest-rest.ons.aliyun.com ";

string topic, secretKey, accessKey;
public MQHelper( string topic, string secretKey, string accessKey)
{
this.topic = topic;
this.secretKey = secretKey;
this.accessKey = accessKey;

}
/// <summary>
/// URL 中的 Key,Tag以及 POST Content-Type 没有任何的限制,只要确保Key 和 Tag 相同唯一即可
/// </summary>
/// <param name="tag"></param>
/// <param name="key"></param>
/// <param name="body"></param>
/// <returns></returns>
public async Task< bool> Pub( string tag, string key, string body)
{
using (HttpClient httpClient = new HttpClient())
{
httpClient.DefaultRequestHeaders.Connection.Add( " keep-alive ");
HttpContent content = new StringContent(body, Encoding.UTF8);
httpClient.DefaultRequestHeaders.Accept.Add( new MediaTypeWithQualityHeaderValue( " text/html "));

var time = ( long)(DateTime.Now.ToUniversalTime() - new DateTime( 1970, 1, 1)).TotalMilliseconds;
var signString = Sign( string.Format( " {0}\nPID_{0}\n{1}\n{2} ", topic, MD5Encrypt(body), time), secretKey);

httpClient.DefaultRequestHeaders.Add( " AccessKey ", accessKey);
httpClient.DefaultRequestHeaders.Add( " Signature ", signString);
httpClient.DefaultRequestHeaders.Add( " ProducerID ", string.Format( " PID_{0} ", topic));

var url = URL + " /message/?topic= " + topic + " &time= " + time + " &tag= " + tag + " &key= " + key;
var res = await httpClient.PostAsync(url, content);
if (res.StatusCode == System.Net.HttpStatusCode.Created)
{
return true;
}
return false;
}
}

public async void Subscribe( string tag = " * ")
{
using (HttpClient httpClient = new HttpClient())
{
httpClient.DefaultRequestHeaders.Connection.Add( " keep-alive ");
httpClient.DefaultRequestHeaders.Add( " Accept-Charset ", " utf-8 ");

var time = ( long)(DateTime.Now.ToUniversalTime() - new DateTime( 1970, 1, 1)).TotalMilliseconds;
var signString = Sign( string.Format( " {0}\nCID_{0}\n{1} ", topic, time), secretKey);

httpClient.DefaultRequestHeaders.Add( " AccessKey ", accessKey);
httpClient.DefaultRequestHeaders.Add( " Signature ", signString);
httpClient.DefaultRequestHeaders.Add( " ConsumerID ", string.Format( " CID_{0} ", topic));

var url = URL + " /message/?topic= " + topic + " &time= " + time + " &num=32 ";
var res = httpClient.GetAsync(url).GetAwaiter().GetResult();
Console.WriteLine(res.StatusCode);
if (res.StatusCode == System.Net.HttpStatusCode.OK)
{
var msg = await res.Content.ReadAsStringAsync();
Console.WriteLine(msg);
if (msg != null && msg.Length > 10)
{
MQMessage[] mqMsgs = JsonConvert.DeserializeObject<MQMessage[]>(msg);
foreach ( var mqMsg in mqMsgs)
{
Delete(mqMsg.msgHandle);
}
}
}
}
}

async void Delete( string msgHandle)
{
using (HttpClient httpClient = new HttpClient())
{
httpClient.DefaultRequestHeaders.Accept.Add( new MediaTypeWithQualityHeaderValue( " text/html "));

var time = ( long)(DateTime.Now.ToUniversalTime() - new DateTime( 1970, 1, 1)).TotalMilliseconds;
var signString = Sign( string.Format( " {0}\nCID_{0}\n{1}\n{2} ", topic, msgHandle, time), secretKey);

httpClient.DefaultRequestHeaders.Add( " AccessKey ", accessKey);
httpClient.DefaultRequestHeaders.Add( " Signature ", signString);
httpClient.DefaultRequestHeaders.Add( " ConsumerID ", string.Format( " CID_{0} ", topic));

var url = URL + " /message/?topic= " + topic + " &time= " + time + " &msgHandle= " + msgHandle;
var res = await httpClient.DeleteAsync(url);
if (res.StatusCode == System.Net.HttpStatusCode.NoContent)
{
Console.WriteLine( " 消息删除成功,无需返回内容 ");
}
else
{
Console.WriteLine(res.StatusCode);
}
}
}

string MD5Encrypt( string strText)
{
using ( var md5 = MD5.Create())
{
var result = md5.ComputeHash(Encoding.UTF8.GetBytes(strText));
return BitConverter.ToString(result).Replace( " - ", "").ToLower();
}
}

string Sign( string signatureString, string secretKey, bool isRaw = true)
{
var enc = Encoding.UTF8;
HMACSHA1 hmac = new HMACSHA1(enc.GetBytes(secretKey));
hmac.Initialize();

byte[] buffer = enc.GetBytes(signatureString);
if (isRaw)
{
byte[] ret = hmac.ComputeHash(buffer);
return Convert.ToBase64String(ret);
}
else
{
string res = BitConverter.ToString(hmac.ComputeHash(buffer)).Replace( " - ", "").ToLower();
return Convert.ToBase64String(Encoding.UTF8.GetBytes(res));
}
}
}

public class MQMessage
{
public string body;
public string bornTime;
public string msgHandle;
public string msgId;
public long reconsumeTimes;
public string tag;
}
复制代码

使用:
复制代码
using MQWebCore;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApp1
{
public class Program
{
public static void Main( string[] args)
{
Console.OutputEncoding = System.Text.Encoding.UTF8;
// Encoding.RegisterProvider(CodePagesEncodingProvider.Instance);
MQHelper mqHelper = new MQHelper( " Test ", "3412qsd's12 ", "3412341212 ");
var res = mqHelper.Pub("testTag", "testKey", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss") + "阿特斯地方").GetAwaiter().GetResult();
Debug.WriteLine(res);
while ( true)
{
mqHelper.Subscribe();
Thread.Sleep( 1000);
}
Console.Read();

}
}
}
复制代码



本文转自94cool博客园博客,原文链接:http://www.cnblogs.com/94cool/p/5747971.html,如需转载请自行联系原作者
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
27天前
|
消息中间件 存储 Serverless
【实践】快速学会使用阿里云消息队列RabbitMQ版
云消息队列 RabbitMQ 版是一款基于高可用分布式存储架构实现的 AMQP 0-9-1协议的消息产品。云消息队列 RabbitMQ 版兼容开源 RabbitMQ 客户端,解决开源各种稳定性痛点(例如消息堆积、脑裂等问题),同时具备高并发、分布式、灵活扩缩容等云消息服务优势。
66 2
|
27天前
|
存储 开发框架 JSON
ASP.NET Core OData 9 正式发布
【10月更文挑战第8天】Microsoft 在 2024 年 8 月 30 日宣布推出 ASP.NET Core OData 9,此版本与 .NET 8 的 OData 库保持一致,改进了数据编码以符合 OData 规范,并放弃了对旧版 .NET Framework 的支持,仅支持 .NET 8 及更高版本。新版本引入了更快的 JSON 编写器 `System.Text.UTF8JsonWriter`,优化了内存使用和序列化速度。
|
2月前
|
消息中间件 弹性计算 运维
阿里云云消息队列RabbitMQ实践解决方案评测报告
阿里云云消息队列RabbitMQ实践解决方案评测报告
70 9
|
2月前
|
开发框架 监控 前端开发
在 ASP.NET Core Web API 中使用操作筛选器统一处理通用操作
【9月更文挑战第27天】操作筛选器是ASP.NET Core MVC和Web API中的一种过滤器,可在操作方法执行前后运行代码,适用于日志记录、性能监控和验证等场景。通过实现`IActionFilter`接口的`OnActionExecuting`和`OnActionExecuted`方法,可以统一处理日志、验证及异常。创建并注册自定义筛选器类,能提升代码的可维护性和复用性。
|
2月前
|
开发框架 .NET 中间件
ASP.NET Core Web 开发浅谈
本文介绍ASP.NET Core,一个轻量级、开源的跨平台框架,专为构建高性能Web应用设计。通过简单步骤,你将学会创建首个Web应用。文章还深入探讨了路由配置、依赖注入及安全性配置等常见问题,并提供了实用示例代码以助于理解与避免错误,帮助开发者更好地掌握ASP.NET Core的核心概念。
76 3
|
17天前
|
开发框架 JavaScript 前端开发
一个适用于 ASP.NET Core 的轻量级插件框架
一个适用于 ASP.NET Core 的轻量级插件框架
|
27天前
|
API
使用`System.Net.WebClient`类发送HTTP请求来调用阿里云短信API
使用`System.Net.WebClient`类发送HTTP请求来调用阿里云短信API
20 0
|
2月前
|
开发框架 NoSQL .NET
利用分布式锁在ASP.NET Core中实现防抖
【9月更文挑战第5天】在 ASP.NET Core 中,可通过分布式锁实现防抖功能,仅处理连续相同请求中的首个请求,其余请求返回 204 No Content,直至锁释放。具体步骤包括:安装分布式锁库如 `StackExchange.Redis`;创建分布式锁服务接口及其实现;构建防抖中间件;并在 `Startup.cs` 中注册相关服务和中间件。这一机制有效避免了短时间内重复操作的问题。
|
2月前
|
监控 安全 搜索推荐
设置 HTTPS 协议以确保数据传输的安全性
设置 HTTPS 协议以确保数据传输的安全性
|
25天前
|
安全 网络协议 算法
HTTPS网络通信协议揭秘:WEB网站安全的关键技术
HTTPS网络通信协议揭秘:WEB网站安全的关键技术
128 4
HTTPS网络通信协议揭秘:WEB网站安全的关键技术

相关产品

  • 云消息队列 MQ