`
wbj0110
  • 浏览: 1598228 次
  • 性别: Icon_minigender_1
  • 来自: 上海
文章分类
社区版块
存档分类
最新评论

RocketMQ入门（2）最佳实践

RocketMQ 
阅读更多

一、服务端安装部署

我是在虚拟机中的CentOS6.5中进行部署。
2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解压到适当的目录如/opt/目录
3.启动RocketMQ：进入rocketmq/bin 目录 执行
<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
nohup sh mqnamesrv&
<!-- [Format Time: 0.0011 seconds] -->
4.启动Broker，设置对应的NameServer
<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
nohup sh  mqbroker-n"127.0.0.1:9876"&
<!-- [Format Time: 0.0010 seconds] -->

 

二、编写客户端

可以查看sameple中的quickstart源码 1.Consumer 消息消费者

<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
/**
* Consumer，订阅消息
*/
publicclassConsumer{
 
    publicstaticvoidmain(String[]args)throwsInterruptedException,MQClientException{
        DefaultMQPushConsumer consumer=newDefaultMQPushConsumer("QuickStartConsumer");
        
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setInstanceName("QuickStartConsumer");
        consumer.subscribe("QuickStart","*");
        
        consumer.registerMessageListener(newMessageListenerConcurrently(){
 
            @Override
            publicConsumeConcurrentlyStatus consumeMessage(List<MessageExt>msgs,
                    ConsumeConcurrentlyContext context){
                System.out.println(Thread.currentThread().getName()+" Receive New Messages: "+msgs);
                returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
 
        consumer.start();
 
        System.out.println("Consumer Started.");
    }
}
<!-- [Format Time: 0.0092 seconds] -->

2.Producer消息生产者

<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
* Producer，发送消息
*
*/
publicclassProducer{
    publicstaticvoidmain(String[]args)throwsMQClientException,InterruptedException{
        DefaultMQProducer producer=newDefaultMQProducer("QuickStartProducer");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setInstanceName("QuickStartProducer");
        producer.start();
  
        for(inti=0;i<1000;i++){
            try{
                Message msg=newMessage("QuickStart",// topic
                    "TagA",// tag
                    ("Hello RocketMQ ,QuickStart"+i).getBytes()// body
                        );
                SendResult sendResult=producer.send(msg);
                System.out.println(sendResult);
            }
            catch(Exceptione){
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
  
        producer.shutdown();
    }
}
<!-- [Format Time: 0.0101 seconds] -->

3.首先运行Consumer程序，一直在运行状态接收服务器端推送过来的消息

<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
23:18:07.587[main]DEBUGi.n.c.MultithreadEventLoopGroup--Dio.netty.eventLoopThreads:16
23:18:07.591[main]DEBUGi.n.util.internal.PlatformDependent-Platform:Windows
23:18:07.592[main]DEBUGi.n.util.internal.PlatformDependent-Java version:7
23:18:07.592[main]DEBUGi.n.util.internal.PlatformDependent--Dio.netty.noUnsafe:false
23:18:07.593[main]DEBUGi.n.util.internal.PlatformDependent0-java.nio.ByteBuffer.cleaner:available
23:18:07.593[main]DEBUGi.n.util.internal.PlatformDependent0-java.nio.Buffer.address:available
23:18:07.593[main]DEBUGi.n.util.internal.PlatformDependent0-sun.misc.Unsafe.theUnsafe:available
23:18:07.593[main]DEBUGi.n.util.internal.PlatformDependent0-sun.misc.Unsafe.copyMemory:available
23:18:07.593[main]DEBUGi.n.util.internal.PlatformDependent0-java.nio.Bits.unaligned:true
23:18:07.594[main]DEBUGi.n.util.internal.PlatformDependent-sun.misc.Unsafe:available
23:18:07.594[main]DEBUGi.n.util.internal.PlatformDependent--Dio.netty.noJavassist:false
23:18:07.594[main]DEBUGi.n.util.internal.PlatformDependent-Javassist:unavailable
23:18:07.594[main]DEBUGi.n.util.internal.PlatformDependent-You don't have Javassist in your class path or you don'thave enough permission toload dynamically generated classes.  Please check the configuration forbetter performance.
23:18:07.595[main]DEBUGi.n.util.internal.PlatformDependent--Dio.netty.noPreferDirect:false
23:18:07.611[main]DEBUG io.netty.channel.nio.NioEventLoop--Dio.netty.noKeySetOptimization:false
23:18:07.611[main]DEBUG io.netty.channel.nio.NioEventLoop--Dio.netty.selectorAutoRebuildThreshold:512
23:18:08.355[main]DEBUGi.n.util.internal.ThreadLocalRandom--Dio.netty.initialSeedUniquifier:0x8c0d4793e5820c31
23:18:08.446[NettyClientWorkerThread_1]DEBUG io.netty.util.ResourceLeakDetector--Dio.netty.noResourceLeakDetection:false
Consumer Started.
<!-- [Format Time: 0.0293 seconds] -->

4.再次运行Producer程序，生成消息并发送到Broker，Producer的日志冲没了，但是可以看到Broker推送到Consumer的一条消息

<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
ConsumeMessageThread-QuickStartConsumer-3Receive NewMessages:[MessageExt[queueId=0,storeSize=150,queueOffset=244,sysFlag=0,bornTimestamp=1400772029972,bornHost=/10.162.0.7:54234,storeTimestamp=1400772016017,storeHost=/127.0.0.1:10911,msgId=0A0A0A5900002A9F0000000000063257,commitLogOffset=406103,bodyCRC=112549959,reconsumeTimes=0,preparedTransactionOffset=0,toString()=Message[topic=QuickStart,flag=0,properties={TAGS=TagA,WAIT=true,MAX_OFFSET=245,MIN_OFFSET=0},body=29]]]
<!-- [Format Time: 0.0068 seconds] -->

 

三、Consumer最佳实践

1.消费过程要做到幂等（即消费端去重）

RocketMQ无法做到消息重复，所以如果业务对消息重复非常敏感，务必要在业务层面去重，有以下一些方式：

（1）.将消息的唯一键，可以是MsgId，也可以是消息内容中的唯一标识字段，例如订单ID，消费之前判断是否在DB或Tair（全局KV存储）中存在，如果不存在则插入，并消费，否则跳过。（实践过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过） msgid一定是全局唯一的标识符，但是可能会存在同样的消息有两个不同的msgid的情况（有多种原因），这种情况可能会使业务上重复，建议最好使用消息体中的唯一标识字段去重

（2）.使业务层面的状态机去重

2.批量方式消费

如果业务流程支持批量方式消费，则可以很大程度上的提高吞吐量，可以通过设置Consumer的consumerMessageBatchMaxSize参数，默认是1，即一次消费一条参数

3.跳过非重要的消息

发生消息堆积时，如果消费速度一直跟不上发送速度，可以选择丢弃不重要的消息

<!-- Crayon Syntax Highlighter v2.6.0 -->
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Override
           publicConsumeConcurrentlyStatus consumeMessage(List<MessageExt>msgs,
                   ConsumeConcurrentlyContext context){
               System.out.println(Thread.currentThread().getName()+" Receive New Messages: "+msgs);
                
               longoffset=msgs.get(0).getQueueOffset();
                
               StringmaxOffset=msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
               longdiff=Long.parseLong(maxOffset)-offset;
               if(diff>100000){
                   //处理消息堆积情况
                   returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               }
                
               returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           }
<!-- [Format Time: 0.0076 seconds] -->

如以上代码所示，当某个队列的消息数堆积到 100000 条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度

4.优化没条消息消费过程

举例如下，某条消息的消费过程如下

1.  根据消息从 DB 查询数据 1

2.  根据消息从 DB 查询数据2

3.  复杂的业务计算

4.  向 DB 插入数据3

5.  向 DB 插入数据 4

这条消息的消费过程与 DB 交互了 4 次，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，也就是说总体性能提高了 40%。

对于 Mysql 等 DB，如果部署在磁盘，那么与 DB 进行交互，如果数据没有命中 cache，每次交互的 RT 会直线上升， 如果采用 SSD，则 RT 上升趋势要明显好于磁盘。

个别应用可能会遇到这种情况：在线下压测消费过程中，db 表现非常好，每次 RT 都很短，但是上线运行一段时间，RT 就会变长，消费吞吐量直线下降

主要原因是线下压测时间过短，线上运行一段时间后，cache 命中率下降，那么 RT 就会增加。建议在线下压测时，要测试足够长时间，尽可能模拟线上环境，压测过程中，数据的分布也很重要，数据不同，可能 cache 的命中率也会完全不同

四、Producer最佳实践

1.发送消息注意事项

（1） 一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。

（2）每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过 topic，key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

（3）消息发送成功或者失败，要打印消息日志，务必要打印 sendresult 和 key 字段

（4）send 消息方法，只要不抛异常，就代表发送成功。但是发送成功会有多个状态，在 sendResult 里定义

  • SEND_OK：消息发送成功
  • FLUSH_DISK_TIMEOUT：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  • ​FLUSH_SLAVE_TIMEOUT：消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  • SLAVE_NOT_AVAILABLE：消息发送成功，但是此时 slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失。对于精确发送顺序消息的应用，由于顺序消息的局限性，可能会涉及到主备自动切换问题，所以如果sendresult 中的 status 字段不等于 SEND_OK，就应该尝试重试。对于其他应用，则没有必要这样

（5）对于消息不可丢失应用，务必要有消息重发机制

2.消息发送失败处理

Producer 的 send 方法本身支持内部重试，重试逻辑如下：

（1） 至多重试 3 次

（2）  如果发送失败，则轮转到下一个 Broker

（3）  这个方法的总耗时时间不超过 sendMsgTimeout 设置的值，默认 10s所以，如果本身向 broker 发送消息产生超时异常，就不会再做重试

如果调用 send 同步方法发送失败，则尝试将消息存储到 db，由后台线程定时重试，保证消息一定到达 Broker。

上述 db 重试方式为什么没有集成到 MQ 客户端内部做，而是要求应用自己去完成，基于以下几点考虑：

（1）MQ 的客户端设计为无状态模式，方便任意的水平扩展，且对机器资源的消耗仅仅是 cpu、内存、网络

（2）如果 MQ 客户端内部集成一个 KV 存储模块，那么数据只有同步落盘才能较可靠，而同步落盘本身性能开销较大，所以通常会采用异步落盘，又由于应用关闭过程不受 MQ 运维人员控制，可能经常会发生 kill  -9 这样暴力方式关闭，造成数据没有及时落盘而丢失

（3）Producer 所在机器的可靠性较低，一般为虚拟机，不适合存储重要数据。 综上，建议重试过程交由应用来控制。

3.选择 oneway 形式发送

一个 RPC 调用，通常是这样一个过程

（1）客户端发送请求到服务器

（2）服务器处理该请求

（3）服务器向客户端返回应答

所以一个 RPC 的耗时时间是上述三个步骤的总和，而某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用 oneway 形式调用，oneway 形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个 os 系统调用的开销，即将数据写入客户端的 socket 缓冲区，此过程耗时通常在微秒级

原文地址：http://www.changeself.net/archives/rocketmq%e5%85%a5%e9%97%a8%ef%bc%882%ef%bc%89%e6%9c%80%e4%bd%b3%e5%ae%9e%e8%b7%b5.html

 

分享到：
| mmap详解
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    阿里RocketMQ_用户指南_V3.2.4 & RocketMQ-原理简介

    阿里RocketMQ是一款由阿里巴巴开源的分布式消息中间件，它在支撑大规模...通过这些文档，读者可以全面了解RocketMQ的原理、使用方法以及最佳实践，从而在实际项目中有效地利用RocketMQ来构建稳定可靠的分布式系统。

    rocketmq小示例项目及Linux下的编译安装说明

    在"RocketMQ学习笔记(1)_Linux下安装RocketMQ.pdf"中，应该详细阐述了这些步骤，并可能包括一些最佳实践和常见问题解决方案。 项目中的`.classpath`和`.project`文件是Eclipse IDE的项目配置文件，它们定义了项目的...

    rocketmq消息中间件.zip

    5. **最佳实践** - **消费策略选择**：理解广播消费与集群消费的适用场景。 - **消息幂等性**：设计幂等性消费者，防止重复消息处理。 - **容错机制**：利用 RocketMQ 提供的特性构建健壮的系统。 - **调优指南*...

    RocketMQ 的 jar 文件 开发文档等资源

    - **最佳实践**：提供了实际应用场景中的经验和建议，帮助避免常见问题。 - **故障排查**：针对可能出现的问题，提供了排查步骤和解决方案。 5. **RocketMQ 扩展功能** - **延时消息**：RocketMQ 支持设置消息...

    rocketmq工具以及官方源码4.2.0版本

    2. 文档涵盖了快速入门、核心概念、设计模式、最佳实践等多个方面，对于开发者来说非常实用。 综上所述，RocketMQ 4.2.0版本不仅提供了强大的消息处理能力，还有丰富的工具和详尽的官方文档支持。通过深入学习源码...

    apache-rocketmq-on-aws：此解决方案可轻松在aws帐户上设置Apache Rocketmq ec2集群

    AWS云上的Apache RocketMQ 此解决方案提供一键式cloudformation部署，以在... 有关体系结构的详细信息，最佳实践，分步说明和自定义选项，请参阅。 通过CloudFormation进行部署 部署时间：大约15分钟。 请按照本节中

    阿里云 专有云企业版 V3.12.0 消息队列 RocketMQ 版 MQTT用户指南 20200623.pdf

    文档可能包含最佳实践指导，比如如何优化消息发布订阅的性能，如何设计合理的Topic结构，以及如何在大规模部署中管理MQTT连接。 **八、版本更新与支持** 由于产品会持续迭代和升级，用户需要定期检查最新的文档版本...

    ROCKET资料.rar

    【压缩包子文件的文件名称列表】: "资料" 这个条目表明压缩包内可能包含多个子目录或文件，如 PDF 文档、源代码文件、Markdown 文档等，这些都可能详细讲解 ROCKET 的核心概念、设计模式、配置选项、最佳实践以及...

    ELK项目前台后系统搭建及Dubbo快速入门.docx

    综上，好客租房项目是一个结合了前沿技术和实际业务场景的应用，其架构设计和实施展示了现代互联网应用的开发趋势和最佳实践。通过ELK和Dubbo的集成，项目不仅实现了高效的数据处理和微服务架构，还提供了强大的日志...

    资料-springcloud alibaba

    在学习资料中，笔记文档会详细介绍这些组件的原理、使用方法和最佳实践，帮助读者快速理解和掌握。源码部分则可以让你深入到每个组件的内部，了解其实现细节。工具可能包括一些辅助开发和调试的工具，如服务治理平台...

    藏经阁-100本阿里系电子书合集-100.pdf

    《大数据工程师必读手册》、《领军行业大数据及AI实战》、《Apache Flink十大技术难点实战》和《Apache Flink年度最佳实践》等书籍，提供了大数据处理和机器学习的深度实践和案例分析。此外，还有《阿里巴巴大数据及...

    Spring Cloud Alibaba 从入门到实战 中文PDF完整版【训练营合集】.pdf

    通过学习Spring Cloud Alibaba，你可以提升构建云原生应用的能力，掌握微服务设计和实施的最佳实践。此外，了解云原生技术如容器（Docker）、服务网格（Istio）等相关知识，将有助于你更好地理解和部署Spring Cloud ...

    spring-cloud-alibaba,spring cloud alibaba为阿里中间件的分布式解决方案提供了一站式的应用开发解决方案。.zip

    - **最佳实践**: 学习社区中的最佳实践案例，优化自己的应用设计和架构。 - **版本更新**: 保持对新版本的关注，以便及时利用新特性并解决潜在问题。 5. **社区支持** - **文档齐全**: Spring Cloud Alibaba提供...

Magicbox
Global site tag (gtag.js) - Google Analytics