Press "Enter" to skip to content

Kettle实现rabbitMQ的生产与消费

本站内容均来自兴趣收集,如不慎侵害的您的相关权益,请留言告知,我们将尽快删除.谢谢.

文章目录

一、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中使用的就是这个路由键。

 

参考文章:

 

使用RabbitMQ搭建MQTT服务

 

Kettle整合Rabbitmq消息队列

基于RabbitMQ在Kettle中使用MQTT协议进行数据传输

三、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组件来进行后续开发,转换命名最好使用英文命名。

Be First to Comment

发表回复

您的电子邮箱地址不会被公开。