01

02

03

RocketMQ 发展历程图

RocketMQ从2017年开始进入Apache孵化，2018年RocketMQ 4.0发布完成云原生化，2021年RocketMQ 5.0发布全面融合消息、事件、流。

1、业务消息领域首选

RocketMQ作为一款“让人睡得着觉的消息产品”成为业务消息领域的首选，这主要源于产品的以下特点：

（1）金融级高可靠

经历了阿里巴巴双十一的洪峰检验；

（2）极简架构

如下图所示， RocketMQ的架构主要包含两部分包括：源数据集群NameServer Cluster和计算存储集群Broker Cluster。

搭建方式：只需要简单的代码就可以搭建RocketMQ集群：

Jar：

 nohup sh bin/mqnamesrv & nohup sh bin/mqbroker -n localhost:9876 &

On K8S：

kubectl apply -f example/rocketmq_cluster.yaml

（3）极低运维成本

RocketMQ的运维成本很低，提供了很好的CLI工具MQAdmin，MQAdmin提供了丰富的命令支持，覆盖集群健康状态检查、集群进出流量管控等多个方面。例如，mqadmin clusterList一条命令可以获取到当前集群全部节点状态（生产消费流量、延迟、排队长度、磁盘水位等）；mqadmin updateBrokerConfig命令可以实时设置broker节点或topic的可读可写状态，从而可以动态摘除临时不可用节点，达到生产消费的流量迁移效果。

（4）丰富的消息类型

RocketMQ支持的消息类型包括：普通消息、事务消息、延迟消息、定时消息、顺序消息等。能够轻松支持大数据场景和业务场景。

（5）高吞吐、低延迟

压测场景主备同步复制模式，每台Broker节点都可以将磁盘利用率打满，同时可以将p99延迟控制在毫秒级别。

2、RocketMQ 5.0概况

RocketMQ 5.0是生于云、长于云的云原生消息、事件、流超融合平台，它具有以下特点：

（1）轻量级SDK

• 全面支持云原生通信标准 gRPC 协议；

• 无状态 Pop 消费模式，多语言友好，易集成；

（2）极简架构

• 无外部依赖，降低运维负担；

• 节点间松散耦合，任意服务节点可随时迁移；

（3）可分可合的存储计算分离

• Broker 升级为真正的无状态服务节点，无 binding；

• Broker 和 Store节点分离部署、独立扩缩；

• 多协议标准支持，无厂商锁定；

• 可分可合，适应多种业务场景，降低运维负担；

  
  

如下图所示，计算集群（Broker）主要包括抽象模型和相对应的协议适配，以及消费能力和治理能力。存储集群（Store）主要分为消息存储CommitLog（多类型消息存储、多模态存储）和索引存储Index（多元索引）两部分，如果可以充分发挥云上存储的能力，将CommitLog和Index配置在云端的文件系统就可以天然的实现存储和计算分离。

（4）多模存储支持

• 满足不同基础场景下的高可用诉求；

• 充分利用云上基础设施，降低成本；

（5）云原生基础设施：

• 可观测性能力云原生化，OpenTelemetry 标准化；

• Kubernetes 一键式部署扩容交付。

04

Hudi 是一个流式数据湖平台，支持对海量数据快速更新。内置表格式，支持事务的存储层、一系列表服务、数据服务（开箱即用的摄取工具）以及完善的运维监控工具。Hudi 可以将存储卸载到阿里云上的 OSS、AWS 的S3这些存储上。

Hudi的特性包括：

• 事务性写入，MVCC/OCC并发控制；

• 对记录级别的更新、删除的原生支持；

• 面向查询优化：小文件自动管理，针对增量拉取优化的设计，自动压缩、聚类以优化文件布局；

05

rocketmq-connect-hudi version：0.0.1-SNAPSHOT

/data/lakehouse/rocketmq-externals/rocketmq-connect/rocketmq-connect-runtime/target/distribution/conf/connect.conf 中connector-plugin 路径

cd rocketmq-externals/rocketmq-connect-hudimvn clean install -DskipTest -U

cd /data/lakehouse/rocketmq-externals/rocketmq-connect/rocketmq-connect-runtime  sh ./run_worker.sh    ##  Worker可以启动多个

curl http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-hudi-sink-connector-name} ?config='{"connector-class":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"topicc","tablePath":"file:///tmp/hudi_connector_test","tableName":"hudi_connector_test_table","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","source-record-converter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","source-rocketmq":"127.0.0.1:9876","src-cluster":"DefaultCluster","refresh-interval":"10000","schemaPath":"/data/lakehouse/config/user.avsc"\}’  启动成功会打印如下日志：2021-09-06 16:23:14 INFO pool-2-thread-1 - Open HoodieJavaWriteClient successfully

cd /Users/osgoo/Documents/hadoop-2.10.1vi core-site.xml<configuration><property> <name>fs.defaultFS</name> <!-- 可以通过命令hostname 查看主机名字  这里的主机名字是hadoop1--> <value>hdfs://localhost:9000</value></property><!--覆盖掉core-default.xml中的默认配置--></configuration>
vi hdfs-site.xml<configuration><property>        <name>dfs.replication</name>        <value>1</value>  </property></configuration>
./bin/hdfs namenode -format./sbin/start-dfs.sh jps 看下namenode,datanodelsof -i:9000./bin/hdfs dfs -mkdir -p /Users/osgoo/Downloads
1、启动rocketmq集群，创建rocketmq-connector内置topicQickStart：https://rocketmq.apache.org/docs/quick-start/sh mqadmin updatetopic -t connector-cluster-topic -n localhost:9876 -c DefaultClustersh mqadmin updatetopic -t connector-config-topic -n localhost:9876 -c DefaultClustersh mqadmin updatetopic -t connector-offset-topic -n localhost:9876 -c DefaultClustersh mqadmin updatetopic -t connector-position-topic -n localhost:9876 -c DefaultCluster
2、创建数据入湖的源端topic，testhudi1sh mqadmin updatetopic -t testhudi1 -n localhost:9876 -c DefaultCluster
3、编译rocketmq-connect-hudi-0.0.1-SNAPSHOT-jar-with-dependencies.jarcd rocketmq-connect-hudimvn clean install -DskipTest -U
4、启动rocketmq-connector runtime配置connect.conf--------------workerId=DEFAULT_WORKER_1storePathRootDir=/Users/osgoo/Downloads/storeRoot
## Http port for user to access REST APIhttpPort=8082
# Rocketmq namesrvAddrnamesrvAddr=localhost:9876
# Source or sink connector jar file dir,The default value is rocketmq-connect-samplepluginPaths=/Users/osgoo/Downloads/connector-plugins---------------拷贝 rocketmq-hudi-connector.jar 到 pluginPaths=/Users/osgoo/Downloads/connector-plugins
sh run_worker.sh
5、配置入湖configcurl http://localhost:8082/connectors/rocketmq-connect-hudi?config='\{"connector-class":"org.apache.rocketmq.connect.hudi.connector.HudiSinkConnector","topicNames":"testhudi1","tablePath":"hdfs://localhost:9000/Users/osgoo/Documents/base-path7","tableName":"t7","insertShuffleParallelism":"2","upsertShuffleParallelism":"2","deleteParallelism":"2","source-record-converter":"org.apache.rocketmq.connect.runtime.converter.RocketMQConverter","source-rocketmq":"127.0.0.1:9876","source-cluster":"DefaultCluster","refresh-interval":"10000","schemaPath":"/Users/osgoo/Downloads/user.avsc"\}'
6、发送消息到testhudi1

7、## 利用spark读取
cd /Users/osgoo/Downloads/spark-3.1.2-bin-hadoop3.2/bin
./spark-shell \  --packages org.apache.hudi:hudi-spark3-bundle_2.12:0.9.0,org.apache.spark:spark-avro_2.12:3.0.1 \  --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'
import org.apache.hudi.QuickstartUtils._import scala.collection.JavaConversions._import org.apache.spark.sql.SaveMode._import org.apache.hudi.DataSourceReadOptions._import org.apache.hudi.DataSourceWriteOptions._import org.apache.hudi.config.HoodieWriteConfig._
val tableName = "t7"val basePath = "hdfs://localhost:9000/Users/osgoo/Documents/base-path7"
val tripsSnapshotDF = spark.  read.  format("hudi").  load(basePath + "/*")tripsSnapshotDF.createOrReplaceTempView("hudi_trips_snapshot")
spark.sql("select * from hudi_trips_snapshot").show()