文章目录
- 一、Kettle为什么可以读取流数据？
- 二、rabbitMQ中启动MQTT插件并创建队列和路由键
- 三、Kettle实现rabbitMQ的生产与消费
Kettle是一款非常强大的ETL工具，不仅可以使用图形化界面，还可以处理各种数据，今天记录一下本人使用Kettle中MQTT组件来实现从rabbitMQ中读取流数据，并进行解析和处理。
提示：以下是本篇文章正文内容，下面案例可供参考
一、Kettle为什么可以读取流数据？
首先，本人使用的是Kettle8.2，里面关于流处理的组件有以下几种 (注意Kettle版本，我现在使用的是8.x版本，这里面只有MQTT组件，可以连接rabbitMQ，但之前使用的7.x版本是没有MQTT流处理的，也就是不能处理rabbitMQ中的数据，而9.x版本中已经有rabbitMQ组件了)：
从流中获取数据信息的第一步就是第一个组件“Get records fromstream”，之后会写到，这些流处理包括JMS、Kafka、MQTT。
然后，Kettle其实是不可以直接连接rabbitMQ的，rabbitMQ默认使用amqp协议，但也可以启用MQTT插件，来使用MQTT协议。因此，我们使用Kettle通过MQTT协议步骤来生产和消费rabbitMQ。
二、rabbitMQ中启动MQTT插件并创建队列和路由键
首先使用rabbitMQ自带的控制台输入命令，也可以用windows cd到rabbit目录输入命令。
输入以下命令：
rabbitmq-plugins enable rabbitmq_mqtt 开启 rabbitmq_mqtt 对应端口 1883 rabbitmq-plugins enable rabbitmq_web_mqtt 开启 rabbitmq_web_mqtt 对应端口 15675
因为我们是使用Kettle来连接rabbitMQ，所以使用的是1883端口，切记，只能使用端口1883，开启之后，可以在 http://rabbitMQ的ip地址:15672/#/ web页面查看端口是否开启：
确定端口开启之后，我们在Exchanges模块下面找到amq.topic交换器，点击进去之后，再绑定队列和路由键：
MQTT官方文档中有涉及到MQTT的系统配置，可自行尝试是否可以更改默认配置，本文未涉及：
值得注意的是我们使用的交换器只能是amq.topic，原因是rabbitMQ中的MQTT插件默认配置中只有一个交换器就是amq.topic，然后队列名称也只能是“mqtt-subscription-”开头，路由键名称可以随便设置。但要便于记忆，后续Kettle中使用的就是这个路由键。
三、Kettle实现rabbitMQ的生产与消费
1、生产数据发送给rabbitMQ
使用Kettle组件：生成记录、MQTT producer
值得注意的是，端口号只能是1883，还有就是下面的topic name是填写路由键，不是topic名称，本次绑定在amq.topic交换器下面队列的路由键是routing.update.username，所以这里填写的就是routing.update.username，其他设置默认就好，如果想要知道其他配置的作用，可参考Kettle的 官方文档 。
2、从rabbitMQ消费数据
使用Kettle组件：MQTT consumer
需要注意的还是端口和路由键，还有就是后续处理步骤最好使用英文命名，使用中文有时候会读取失败，或者识别不到XML文件，或者报错不是.ktr文件，重点切记！！！
后续处理步骤使用组件：Get records fromstream、表输出、空操作、写日志、transformation executor
“Get records fromstream”从流中接收信息，“表输出”将接收的信息存储到数据库中，“空操作”插入数据库时如果报错的消极处理，也可以换成“excel输出”，存储报错信息，“写日志”是将接收到信息打印到控制台，“transformation executor”是指定一个子转换步骤来处理数据，如后续没有处理需求，该步骤可省略，可只使用“Get records fromstream”和“写日志”两个步骤就行，进行验证。因为本次处理的数据为Json数据，所以还要对Json数据进行解析和处理，然后再使用解析后的数据去更新相关数据表。
接收到Json数据存储到了Mysql数据库中，所以解析就使用了Mysql自带的函数（JSON_EXTRACT），使用方法可参考文章：
mysql解析json字符串_Mysql解析json字符串/数组
也可参考本人的sql来解析Json数组：
select x.id, x.only_id, x.createby, x.createtime, x.platform, x.shopname, x.realshopname, x.username, x.oldusername, x.rownum, y.user_id from ( select a.id, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].id'),'"','') as only_id, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].createBy'),'"','') as createby, from_unixtime(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].createTime')/1000,'%Y-%m-%d %H:%i:%S') as createtime, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].platform'),'"','') as platform, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].shopName'),'"','') as shopname, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].realShopName'),'"','') as realshopname, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].userName'),'"','') as username, replace(json_extract(substring_index( substring_index( a.message, ";", b.id ), ";",- 1 ), '$[0].oldUserName'),'"','') as oldusername, b.id as rownum from (select id,replace(replace(replace(message,"},{","};{"),"]",""),"[","") as message,flag from sys_update_shopusername_log) a join mysql.help_toplic_autonum b on b.id <= ( length( a.message ) - length( replace ( a.message, ";", "" ) ) + 1 ) where a.flag = 0 ) x join sys_user_detail y on x.username = y.username and upper(x.platform) = y.platform -- “sys_update_shopusername_log”为存储的消费到的Json数据 -- “mysql.help_toplic_autonum”自定义拆分Json数组的辅助表
先启动MQTT消费者，如报错，就检查ip地址、端口、路由键是否正确。启动完成后再启动MQTT生产者，发送消息给rabbitMQ，再自己消费。
消费者：
生产者：
再次查看消费者消费情况：
可以看到是能够生产数据和消费数据，这个之后就可以让上游开发将数据信息发送到我们的默认交换器amq.topic的绑定队列里面，我们就可以消费和处理了。
四、总结
注意细节：是否开启MQTT插件，端口号是否是1883，交换器和队列名称是否符合默认设定，Kettle里面MQTT producer和MQTT consumer组件所涉及到topic name 都是路由键，是在rabbitMQ中创建队列时绑定的路由键，最后就是可以根据接收到消息使用transformation executor组件来进行后续开发，转换命名最好使用英文命名。
提示：如本文有一点点帮助到您，请点赞、转发、收藏、留言，感谢！！！，如需转载、引用敬请注明！！！
