2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解压到适当的目录如/opt/目录
4.启动Broker,设置对应的NameServer
|
nohup sh mqbroker-n"127.0.0.1:9876"&
|
二、编写客户端
可以查看sameple中的quickstart源码 1.Consumer 消息消费者
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.");
}
}
|
2.Producer消息生产者
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();
}
}
|
3.首先运行Consumer程序,一直在运行状态接收服务器端推送过来的消息
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.
|
4.再次运行Producer程序,生成消息并发送到Broker,Producer的日志冲没了,但是可以看到Broker推送到Consumer的一条消息
|
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]]]
|
三、Consumer最佳实践
1.消费过程要做到幂等(即消费端去重)
RocketMQ无法做到消息重复,所以如果业务对消息重复非常敏感,务必要在业务层面去重,有以下一些方式:
(1).将消息的唯一键,可以是MsgId,也可以是消息内容中的唯一标识字段,例如订单ID,消费之前判断是否在DB或Tair(全局KV存储)中存在,如果不存在则插入,并消费,否则跳过。(实践过程要考虑原子性问题,判断是否存在可以尝试插入,如果报主键冲突,则插入失败,直接跳过) msgid一定是全局唯一的标识符,但是可能会存在同样的消息有两个不同的msgid的情况(有多种原因),这种情况可能会使业务上重复,建议最好使用消息体中的唯一标识字段去重
(2).使业务层面的状态机去重
2.批量方式消费
如果业务流程支持批量方式消费,则可以很大程度上的提高吞吐量,可以通过设置Consumer的consumerMessageBatchMaxSize参数,默认是1,即一次消费一条参数
3.跳过非重要的消息
发生消息堆积时,如果消费速度一直跟不上发送速度,可以选择丢弃不重要的消息
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;
}
|
如以上代码所示,当某个队列的消息数堆积到 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
相关推荐
RocketMQ最佳实践
rocketmq入门包括基本概念,环境搭建部署,java操作api
RocketMQ入门实战及源码解析.7z
RocketMQ最佳实践,高清,阿里巴巴官方出品,RocketMQ学习必备,RocketMQ调优必备
本文来自于jianshu,文章介绍了Rocketmq、kafka、Rabbitmq的详细对比,RocketMQ集群部署等相关内容。一、MQ背景&选型 消息队列作为高并发系统的核心组件之一,能够帮助业务系统解构提升开发效率和系统稳定性。主要...
分布式消息引擎Apache RocketMQ最佳实践
一、rocketmq入门到精通视频教程目录大纲 001-001_RocketMQ_简介 002-002_RocketMQ_核心概念详解 003-003_RocketMQ_集群构建模型详解(一) 004-004_RocketMQ_集群构建模型详解(二) 005-005_RocketMQ_双主模式集群...
本文来自于aliyun,文章RocketMQ的特点、网络部署、关键特性,以及Consumer最佳实践等相关内容。RocketMQ是一款分布式、队列模型的消息中间件,具有以下特点:能够保证严格的消息顺序提供丰富的消息拉取模式高效的...
基于DLedger构建高可用RocketMQ集群实践.pdf
想要快速入门rocketMq的均可下载,适合新手小白学习。快速上手
2.RocketMQ入门 2.1.rocketmq安装 2.2.一对一模式 2.3.广播模式 3.消息的特性 3.1.同步消息 3.2.异步消息 3.3.单向消息 3.4.延时消息 3.5.批量发送 3.6.顺序性 3.7.消息事务 3.8.过滤消息tag 4.rocketmq集群 4.1....
内容概要: 了解什么RocketMQ 了解RocketMQ的核心概念 动手安装RocketMQ服务 快速入门,掌握RocketMQ的api使用 对producer、consumer进行详解 了解RocketMQ的存储特点
某马程序员-JavaEE 57期-day09-分布式消息系统RocketMQ的入门 代码、讲义和资料
rocketmq可视化控制台最新版
为了帮助大家更好的学习和使用RocketMQ,因此提供相关学习文档,一起学习,资源包括 RocketMQ初步认知、RocketMQ单机环境搭建、RocketMQ集群部署实践、基于myeclipse的RocketMQ--Demo实践、基于RocketMQ--Demo项目的...
一、rocketmq入门到精通视频教程目录大纲 001-001_RocketMQ_简介 002-002_RocketMQ_核心概念详解 003-003_RocketMQ_集群构建模型详解(一) 004-004_RocketMQ_集群构建模型详解(二) 005-005_RocketMQ_双主模式集群...
1.能够保证严格的消息顺序2.提供丰富的消息拉取模式3.高效的订阅者水平扩展能力4.实时的消息订阅机制5.亿级消息堆积能力(1)NameServer是一个几乎无状态的节点,可集群部署,节点之间无任何信息同步(2)Broker部署...