阿里云消息队列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版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
物联网
(手把手)在华为云、阿里云搭建自己的物联网MQTT消息服务器,免费IOT平台
本文介绍如何在阿里云搭建自己的物联网MQTT消息服务器,并使用 “MQTT客户端调试工具”模拟MQTT设备,接入平台进行消息收发。
3832 42
|
消息中间件 存储 Serverless
【实践】快速学会使用阿里云消息队列RabbitMQ版
云消息队列 RabbitMQ 版是一款基于高可用分布式存储架构实现的 AMQP 0-9-1协议的消息产品。云消息队列 RabbitMQ 版兼容开源 RabbitMQ 客户端,解决开源各种稳定性痛点(例如消息堆积、脑裂等问题),同时具备高并发、分布式、灵活扩缩容等云消息服务优势。
507 2
|
消息中间件 安全 API
《阿里云产品四月刊》—Apache RocketMQ ACL 2.0 全新升级(1)
阿里云瑶池数据库云原生化和一体化产品能力升级,多款产品更新迭代
865 1
《阿里云产品四月刊》—Apache RocketMQ ACL 2.0 全新升级(1)
|
消息中间件 安全 Apache
《阿里云产品四月刊》—Apache RocketMQ ACL 2.0 全新升级(2)
阿里云瑶池数据库云原生化和一体化产品能力升级,多款产品更新迭代
651 0
《阿里云产品四月刊》—Apache RocketMQ ACL 2.0 全新升级(2)
|
监控 物联网 网络性能优化
【杂谈】-MQTT与HTTP在物联网中的比较:为什么MQTT是更好的选择
通过上述分析,可以看出MQTT在物联网应用中的确是更好的选择。其高效的通信模型、低带宽消耗、稳定的连接保持机制以及可靠的消息质量保证,使其在各种物联网场景中都能表现出色。开发者在设计和实现物联网系统时,应优先考虑采用MQTT协议,以充分发挥其在资源受限环境下的优势,提升系统的整体性能和可靠性。
3553 26
|
消息中间件 存储 大数据
阿里云消息队列 Kafka 架构及典型应用场景
阿里云消息队列 Kafka 是一款基于 Apache Kafka 的分布式消息中间件,支持消息发布与订阅模型,满足微服务解耦、大数据处理及实时流数据分析需求。其通过存算分离架构优化成本与性能,提供基础版、标准版和专业版三种 Serverless 版本,分别适用于不同业务场景,最高 SLA 达 99.99%。阿里云 Kafka 还具备弹性扩容、多可用区部署、冷热数据缓存隔离等特性,并支持与 Flink、MaxCompute 等生态工具无缝集成,广泛应用于用户行为分析、数据入库等场景,显著提升数据处理效率与实时性。
|
消息中间件 测试技术
通过轻量消息队列(原MNS)主题HTTP订阅+ARMS实现自定义数据多渠道告警
轻量消息队列(原MNS)以其简单队列模型、轻量化协议及按量后付费模式,成为阿里云产品间消息传输首选。本文通过创建主题、订阅、配置告警集成等步骤,展示了该产品在实际应用中的部分功能,确保消息的可靠传输。
444 2
|
网络协议 物联网 网络性能优化
物联网协议比较 MQTT CoAP RESTful/HTTP XMPP
【10月更文挑战第18天】本文介绍了物联网领域中四种主要的通信协议:MQTT、CoAP、RESTful/HTTP和XMPP,分别从其特点、应用场景及优缺点进行了详细对比,并提供了简单的示例代码。适合开发者根据具体需求选择合适的协议。
779 5
|
消息中间件 弹性计算 运维
阿里云云消息队列RabbitMQ实践解决方案评测报告
阿里云云消息队列RabbitMQ实践解决方案评测报告
332 9
|
网络协议 物联网 网络性能优化
物联网江湖风云变幻!MQTT CoAP RESTful/HTTP XMPP四大门派谁主沉浮?
【9月更文挑战第3天】物联网(IoT)的兴起催生了多种通信协议,如MQTT、CoAP、RESTful/HTTP和XMPP,各自适用于不同场景。本文将对比这些协议的特点、优缺点,并提供示例代码。MQTT轻量级且支持QoS,适合大规模部署;CoAP基于UDP,适用于低功耗网络;RESTful/HTTP易于集成但不适合资源受限设备;XMPP支持双向通信,适合复杂交互应用。通过本文,开发者可更好地选择合适的物联网通信协议。
409 2

相关产品

  • 云消息队列 MQ