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

2022-11-02 08:23:08

　由于公司的最近开始使用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 是一个款非常强大的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调用会有问题.后面换成监听类, 问题就解决了.

码农公寓

特色功能

多协议接入

相关文章