目录
1：RocketMQ生产消息 异步发送
2：RocketMQ OneWay发送消息及多种场景对比
3：RocketMQ 延迟消息实战 定时消息 与在电商系统中应用
4: RocketMQ4.X里面的标签Tag实战和消息过滤原理
5： PushConsumer（常用）、PullConsumer消费模式分析
1：RocketMQ生产消息 异步发送
可用于需求速度非常高的时候
特别注意 最后defaultMQProducer的shutdown之前 需要先睡眠几秒钟
避免异步结果没有回来就关闭了生产者
//生产者同步发送消息
//结果同步返回
SendResult sendResult = defaultMQProducer.send(message);
log.info("RocketMQ order 消息发送结果：{}", JSON.toJSONString(sendResult));
//异步异步发送消息（需求要求速度非常高的时候 ）
//结果异步返回
defaultMQProducer.send(message, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("RocketMQ order 消息异步发送结果：{}", JSON.toJSONString(sendResult));
//发送成功可以触发其他的逻辑
}
@Override
public void onException(Throwable throwable) {
log.info("异步发送返回异常:"+throwable,throwable.getMessage());
}
});
//关闭生产者之前先睡眠几秒钟
TimeUnit.SECONDS.sleep(5);
defaultMQProducer.shutdown();
2：RocketMQ OneWay发送消息及多种场景对比
简介：讲解使用RocketMQ发送oneway消息和使用场景，多种发送模式对比
SYNC :同步（不丢失）
应用场景：重要通知邮件、报名短信通知、营销短信系统等
ASYNC :异步（不丢失）
应用场景：对RT时间敏感,可以支持更高的并发，回调成功触发相对应的业务，比如注册成功 后通知积分系统发放优惠券
ONEWAY :无需要等待响应 （很快 有可能丢失）
使用场景：适用于某些耗时非常短，,只负责发送消息，但对可靠性要求并不高的场景,也不做要求。
主要是日志收集，如日志服务LogServer，该服务作为消费端 通过ONEWAY发送消息过来，里面的队列可以存放消息。
//生产者同步发送消息
SendResult sendResult = defaultMQProducer.send(message);
log.info("RocketMQ order 消息发送结果：{}", JSON.toJSONString(sendResult));
//生产者同步发送OneWay消息
//没有结果的返回
defaultMQProducer.sendOneWay(message);
3：RocketMQ 延迟消息实战 定时消息 与在电商系统中应用
延迟消息：
Producer将消息发送到消息队列RocketMQ服务端，但并不期望这条消息立马投递，
而是推 迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息，
目前 支持固定精度的消息
代码：rocketmq-store > MessageStoreConfig.java 属性 messageDelayLevel
"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
//xxx是级别，1表示配置里面的第一个级别，2表示第 二个级别
message.setDelayTimeLevel(xxx)
定时消息：目前rocketmq开源版本还不支持，商业版本则有，两者使用场景类似
使用场景：
1：消息生产和消费有时间窗口要求：
比如在天猫电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。
这条消息将会在30分钟以后投递绐消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。
如没有支付则取消 如已支付则忽略
2：通过消息触发一些定时任务：
比如在某一固定时间点向用户发送提醒消息（用户会员于到期那天的9点 发送提醒消息）
4: RocketMQ4.X里面的标签Tag实战和消息过滤原理
一个Message只有一个Tag, tag是二级分类
订单：数码类订单、食品类订单
过滤分为Broker端和Consumer端过滤
Consumer端过滤：完全可以根据业务需求进行调整，但是增加了很多无用的消息传输
Broker端过滤：减少了无用的消息的进行网络传输，增加了broker的负担
Consumer端过滤 （tag（常用） 或者 sql（了解））
消息发送端
Message message = new Message(topic,tag, JSON.toJSONString(data).getBytes());
消费端
一般是监听*，
或者指定tag（tag性能高，逻辑简单）
1：defaultMQPushConsumer.subscribe(topic, "*");
2：defaultMQPushConsumer.subscribe(topic, "my_tag");
3：defaultMQPushConsumer.subscribe(topic, "my_tag_A || my_tag_B");
消息发送端:
Message message = new Message(topic,tag, JSON.toJSONString(data).getBytes());
message.putUserProperty("amount",amount)
消费端：
也可以使用SQL92（支持复杂逻辑 只支持PushConsumer中使用MessageSelector.bySql() （了解即可）
获取该Topic下 消息发送端设置了 putUserProperty "amount >5"的所有消息
defaultMQPushConsumer.subscribe(topic,MessageSelector.bySql("amount >5") )
SQL92的常见错误：
The broker does not support consumer to filter message by SQL92
解决：broker.conf里面配置如下
enablePropertyFilter=true
注意：如果想使用多个Tag,可以使用sql表达式，但是不建议，单一职责，多个队列
Broker端过滤（了解）
第一次对比（hashcode）：
在Broker端进行MessageTag过滤，遍历message queue存储的message tag和 Consumer订阅的tag
如果两者的hashcode不一样则跳过。符合的则传输绐Consumer。
第二次对比（tag）
为了防止 通过第一次对比的hashcode 是由于hash碰撞产生的
Consumer收到过滤消息后也会进行匹配操作 ，但是对比真实的message tag而不是hashcode。
优点和保障
1：过滤中不访问commit log（后续提及），可以高效过滤 （优点）
2：consume queue存储使用hashcode定长，节约空间 （优点 保障）
3：如果存在hash冲突，Consumer端可以进行再次确认 （保障）
5： PushConsumer（常用）、PullConsumer消费模式分析
消费端的三种重要概念：
1：Pull
2：Push
3：长轮训
消费端一共有几种处理消息的方式
1：PushConsumer 本质是长轮训
2：PullConsumer 本质是Pull
Pull（很少用）
消费端从Broker拉取消息，主动权在消费者端，可控性好；
但间隔时间不好设置：
时间间隔太短，则有可能什么都拉取不到请求，浪费资源；
时间间隔太长，则消息不能及时处理。
Push（常用）
实时性高；
增加Broker的压力，消费端能力不同，如果Push推送过快，消费端有可能会出现消息堆积。
长轮询（结合pull 与 Push 的优点）：
消费端请求Broker的时候，Broker会保持当前连接一段时间默认是 15s，
如果这段时间内有消息到达，则立刻返回绐Consumer，
如果没消息的时间超过15s,则返回为空，Consumer再进行重新请求；
主动权在Consumer中（因为需要Consumer进行重新请求），Broker即使有大量的消息也不会主动提送 Consumer。
缺点：Broker需要保持Consumer的请求，会占用资源，需要消费端连接数可控 否则会一堆连接
PushConsumer本质是长轮训
DefaultMQPushConsumerImpl
系统收到消息后自动处理消息和offset（后续提及）
在broker端可以通过longPollingEnable=true来开启长轮询（默认开启）
broker端代码（RocketMQ源码）：broker.longpolling
消息消费端代码：DefaultMQPushConsumerImpl->pullMessage->PullCallback
优雅关闭：主要是释放资源和保存Offset,调用shutdown。
虽然方法名是PushConsumer，但是代码里面大量使用了pull，
因为使用长轮训方式：既有Broker的Push的实时性 又有Consumer的Pull的主动性。
Offset的存放（后续会提及）：
集群消费模式存放在Broker端 广播消费模式存放在Consumer端
PullConsumer （了解）（需要自己维护Offset）
DefaultMQPullConsumer
参考官方例子 org.apache.rocketmq.example.simple.PullConsumer
1：进行MessageQueue遍历
2：每一次获取MessageQueue 的消息的时候 获取下一条消息的Offset
3：消费端维护Offset，需用本地存储Offset,存储内存、磁盘、数据库等
4：处理不同状态的消息 4种状态：
FOUND（成功）、NO_NEW_MSG（没有新消息）、 NO_MATCHED_MSG（没有匹配）、OFFSET_ILLRGL（偏移量错误）
优雅关闭：主要是释放资源和保存Offset，需用Consumer端保存好Offset，特别是异常处理的时候
灵活性高可控性强，但是编码复杂度很高
Offset的存放（后续会提及）：
为什么PullConsumer必须把Offset存放在本地：
因为这样可以更方便 请求获取对Offert后 进行各种处理
项目连接
请配合项目代码食用效果更佳：
项目地址：
https://github.com/hesuijin/hesuijin-study-project
Git下载地址：
https://github.com.cnpmjs.org/hesuijin/hesuijin-study-project.git
rocketmq-module项目模块下
注意：因为需要修改相应配置 相关测试的代码
生产者代码主要在 单元测试
消费者代码主要在 项目代码 com.example.rocketmq.demo.consumer.Junit包下
网友评论