［原创］阿里云RocketMQ踩过的哪些坑

时间：2022-10-18 01:45:52

　由于公司的最近开始使用RocketMQ来做支付业务处理, 便开启了学习阿里云RocketMQ的学习与实践之路, 其中踩了不少的坑, 大部份是由于没有仔细查看阿里云的技术文档而踩的坑. 但是有一个非常大的坑, 确实是阿里云的技术文档里没有提及, 在文章的最后会给大家提及.

使用过程了封装的RocketMQ类库已开源在github: https://github.com/antaintan/easyrocketmq

公司的一位同事推荐使用RocketMQ, 并给出了几个流行MQ的对比, 资料来源, 阿里云帮助文档

目前RocketMQ已经成为Apache*项目, 作为开源软件免费提供. 但是阿里云提供的RocketMQ是收费的, 一个Topic 2元/天, 2元100万消息.具体价格,可以参照阿里云的报价.

详细对比资料地址: https://help.aliyun.com/document_detail/52577.htmlhttps://help.aliyun.com/document_detail/52577.html

(注: 阿里云的技术文档写得不错, 可以学习到很多知识)

［原创］阿里云RocketMQ踩过的哪些坑

　　RocketMQ 是一个款非常强大的MQ, 它同时支持, 无序消息, 分区顺序消息, 全局顺序消息, 并支持TCP, Http协议接入.

　　客户端支持Java, C++/.Net(.Net客户端是通过封装C++DLL，通过PInvoke调用来实现的, 基于x64位, 所以我们自己项目编译时也要选择x64)

特色功能

  • 事务消息，实现类似 X/Open XA 的分布事务功能，以达到事务最终一致性状态。
  • 定时（延时）消息，允许消息生产者指定消息进行定时（延时）投递，最长支持40天。
  • 大消息，目前默认支持最大 256KB 消息，华北2 地域支持最大 4MB 消息。
  • 消息轨迹，通过消息轨迹，用户能清晰定位消息从发布者发出，经由 MQ 服务端，投递给消息订阅者的完整链路，方便定位排查问题。
  • 广播消息，允许一个 Consumer ID 所标识的所有 Consumer 都会各自消费某条消息一次。
  • 顺序消息，允许消息消费者按照消息发送的顺序对消息进行消费。
  • 重置消费进度，根据时间重置消费进度，允许用户进行消息回溯或者丢弃堆积消息。

多协议接入

  • 支持 HTTP 协议：支持 RESTful 风格 HTTP 协议完成收发消息,可以解决跨语言使用 MQ 问题。
  • 支持 TCP 协议：区别于 HTTP 简单的接入方式，提供更为专业、可靠、稳定的 TCP 协议的 SDK 接入。

注意: Java客户端的TCP协议是功能支持最全的, .Net TCP要少一些功能支持如顺序消息, 而Http协议支持的更少, 具体情况, 请详细查阅阿里云的帮助文档.

消息队列的几个核心对象:

  • Topic: 消息主题
  • Consumer: 消息消费者, 消费订阅的主题消息
  • Producer: 消息生产者, 生产不同的消息

　　阿里提供的.Net类库几个核心类库:

NSClient4CPP.lib, c++类库

ONSClient4CPP.dll, c++类库

ONSClient4CPP.pdb c++类库

ons.dll: Net类库, 利用开源软件 SWIG 生成 PINVOKE 封装代码

如果系统没有c++运行时类库, 还需要安装vc_redist.x64.exe.

下面是生产者示例代码:

private static ProducerClient producerClient = new ProducerClient(AccessKeyId, AccessKeySecret, ProducerId);

private static void Main(string[] args)

{

    producerClient.Start();

    var stopWatch = new Stopwatch();

    stopWatch.Start();

    var taskList = new List<Task>();

    for (int threadIndex = ; threadIndex <= ProducerThreadCount; threadIndex++)

    {

        // 生产消费

        var task = Task.Factory.StartNew(() => {

            for (int messageIndex = ; messageIndex <= MessageCountPerThread; messageIndex++)

            {

                string content = "线程ID=" + Thread.CurrentThread.ManagedThreadId + ", 我要测试rocketmq message";

                //producerClient.SendMessage(ShardingKey, Topic, content, Tag);

                producerClient.SendMessage(Topic, content, Tag);

                Console.WriteLine(content);

            }

        }, TaskCreationOptions.LongRunning);

        taskList.Add(task);

    }

    Task.WaitAll(taskList.ToArray());

    stopWatch.Stop();

    // 一定要关闭，不然会有内存泄漏

    producerClient.Shutdown();

    Console.WriteLine("发送消息：{0}条， 使用时间{1}毫秒", MessageCountPerThread * ProducerThreadCount, stopWatch.ElapsedMilliseconds);

    Console.ReadLine();

}

下面是消费者示例代码
private static PushConsumerClient consumerClient = new PushConsumerClient(AccessKeyId, AccessKeySecret, Topic, ConsumerId, SubExpression);

private static int count = ;

private class MyMsgListener : DefaultMessageListener

{

    public override ons.Action consume(Message message, ConsumeContext context)

    {

        Console.WriteLine("消息序号: {0}, 当前线程ID = {1}, 内容为： {2}", ++count, Thread.CurrentThread.ManagedThreadId, message.getBody());

        return ons.Action.CommitMessage;

    }

}

private static void Main(string[] args)

{

    var listener = new MyMsgListener();

    consumerClient.setMessageListener(listener);

    consumerClient.Start();

    Console.ReadLine();

    consumerClient.Shutdown();

}

　　消费者有一个消息监听类, 有个consume方法, 这里趟过一个大坑, 因为为了省去一个类文件, 就用了匿名方法来代替上面的监听类, 结果只要消息一条消息后, 程序就自动死, 没有任何.net错误信息, 只有一个很底层的错误, 得不到任何有价值的信息, 花了很多时间, 才发现是因为匿名方法通过PPInvoke调用会有问题.后面换成监听类, 问题就解决了.

相关文章