thingsboard规则链调用外部API——kafka

分类：Kafka作者：Ayu大象

文章目录

thingsboard规则链调用外部API——kafka

因为只是测试一下这个流程，就不做集群了，下面是kafka以及zookeeper的软件安装包

链接：https://pan.baidu.com/s/1VXHAMJ7lKUJvs6Zc6zNKjg?pwd=t5xq 
提取码：t5xq
一、安装zookeeper

因为kafka依赖于zookeeper，首先安装zookeeper，下面是步骤：

（1）拷贝zookeeper安装包到/opt/module下面

（2）解压缩：tar -zxvf 安装包名称，使用 “mv “ 指令可以重命名文件夹

（3）修改配置文件：

#将/opt/module/zookeeper-3.5.7/conf 这个路径下的 zoo_sample.cfg 修改为 zoo.cfg
 mv zoo_sample.cfg zoo.cfg
#打开 zoo.cfg 文件，修改 dataDir 路径：
vim zoo.cfg
dataDir=/opt/module/zookeeper-3.5.7/zkData
#在/opt/module/zookeeper-3.5.7/创建文件夹zkData
mkdir zkData

（4）启动zookeeper：./bin/zkServer.sh start

（5）查看启动状态：./bin/zkServer.sh status

ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Mode: standalone

有时候启动不成功（通过jps也可以查看是否启动成功），看看是不是8080端口被占用了（在logs文件夹下面查看启动日志）,如果是,在zoo.cfg添加一条配置,修改zookeeper的端口：

#serverport
admin.serverPort=你想使用的端口号

（6）zookeeper相关指令：

#启动：
./zkServer.sh start
#查看状态
./zkServer.sh status
#关闭
./zkServer.sh stop
#进入客户端
./zkCli.sh
#推出客户端
quit
#显示客户端所有保存的文件
ls /
#删除某个数据文件夹
deleteall xxx
二、安装kafka

安装步骤，相同的部分省略了：

（1）上传安装包，解压缩

（2）修改配置文件

vi config/server.properties
#修改下面几个参数：
#broker 的全局唯一编号，不能重复，只能是数字。
broker.id=0
#kafka 运行日志(数据)存放的路径，路径不需要提前创建，kafka 自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
log.dirs=/opt/module/kafka/datas
advertised.listeners=PLAINTEXT://服务器ip地址:9092
#配置连接Zookeeper 集群地址（在 zk 根目录下创建/kafka，方便管理） 
#示例：zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka
zookeeper.connect=localhost:2181/kafka

（3）配置环境变量：

vim /etc/profile.d/my_env.sh
增加如下内容:
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka 
export PATH=$PATH:$KAFKA_HOME/bin
#刷新环境变量
source /etc/profile

（4）启动kafka：./bin/kafka-server-start.sh -daemon config/server.properties

（5）停止kafka：./bin/kafka-server-stop.sh

（6）查看启动状态：jps

（7）相关指令：

#1.启动
./bin/kafka-server-start.sh -daemon config/server.properties
#2.停止
./bin/kafka-server-stop.sh 
#3.查看所有的topic：
bin/kafka-topics.sh --bootstrap-server 自己ip:9092 --list
#4.创建主题
bin/kafka-topics.sh --bootstrap-server 自己ip:9092 --create --partitions 1 --replication-factor 3 --topic first
#选项说明：
--topic 定义 topic 名
--replication-factor 定义副本数
--partitions 定义分区数
#5.修改分区数（注意：分区数只能增加，不能减少）
bin/kafka-topics.sh --bootstrap-server 自己ip:9092 --alter --topic first --partitions 3
#6.查看first 主题的详情
bin/kafka-topics.sh --bootstrap-server 自己ip:9092 --describe --topic first
#7.删除主题
bin/kafka-topics.sh --bootstrap-server 自己ip:9092 --delete --topic first
#8.启动消费者
bin/kafka-console-consumer.sh --bootstrap-server 自己ip:9092 --topic first
#9.启动生产者
bin/kafka-console-producer.sh --bootstrap-server 自己ip:9092 --topic first
三、thingsboard配置

（1）准备模拟遥测数据，这是我准备的数据格式：

	{
		"alarmType": "0",
		"deviceAddress": "9",
		"deviceSn": "19_bxj_cqzx_9",
		"directionInfo": "cqzz",
		"highAlarm": "45",
		"id": "19_bxj_cqzx_9",
		"lowAlarm": "5",
		"section": "19",
		"sensorName": "温度",
		"sensorValue": "40",
		"symbols": "temperature",
		"tunnel": "bxj",
		"unitValue": "40 °C"
	}

（2）我用springboot的@Scheduled注解定时发送，我使用http和mqtt两种协议推送数据：

    @Scheduled(cron = "0/5 * * * * ?")
    public void doTask() {
        if (flag) {
            //http数据推送
            String newData = getNewData();
            pushData(newData);

            //mqtt数据推送
            String mqttData = getNewData();
            mqttPushData(mqttData);
        }
    }

     public String getNewData() {

        int max = list.size();
        //生成随机数[a,b],公式：int num = min + (int)(Math.random() * (max-min+1));
        int r = (int) (Math.random() * max);
        return JSONObject.toJSONString(list.get(r));

    }

（3）责任链配置：

在这里插入图片描述

#1.过滤数据:
return msg.sensorName == "氧气";
#2.log
return '推送到kafka数据:\n' + JSON.stringify(msg) + '\n推送到kafka元素据:\n' + JSON.stringify(metadata);

在这里插入图片描述

四、springboot集成kafka消费数据

（1）依赖：

        <!--kafka begin-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!--kafka end-->

（2）application配置

###########【Kafka集群】###########
spring.kafka.bootstrap-servers=ip:9092
###########【初始化生产者配置】###########
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了
​
# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 自定义分区器
# spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner
​
###########【初始化消费者配置】###########
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=defaultConsumerGroup
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=latest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时，项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
# spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
# spring.kafka.consumer.max-poll-records=50

（3）简单消费者

@Component
@Slf4j
public class SimpleConsumer {

    /**
     * 简单消费
     * @param record record
     */
    @KafkaListener(topics = {"tb-devices"})
    public void onMessage(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        log.warn("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
    }
    
}

（4）日志：

在这里插入图片描述

至此，thingboard发送遥测数据到kafka的简单demo已经实现，下面是我参考的文章：

#springboot集成kafka
https://blog.csdn.net/weixin_70730532/article/details/125425798
#kafka安装部署
https://www.jianshu.com/p/61b224cea03b
#thingsboard中文网
http://www.ithingsboard.com/

我的代码地址:

https://gitee.com/ayu-elephant/envmonitoring.git

原文地址：https://blog.csdn.net/qq_38899062/article/details/129139017

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

相关推荐

Kafka的认证
文章浏览阅读4.1k次。kafka认证_kafka认证
Kafka之常用参数配置整理
文章浏览阅读4.8k次，点赞4次，收藏11次。kafka常用参数_kafka配置
Kafka生产者发送消息的流程
文章浏览阅读1.4k次，点赞25次，收藏10次。Kafka 生产者发送消息的流程涉及多个步骤，从消息的创建到成功存储在 Kafka 集群中。_kafka发送消息流程
第3、4章 Kafka 生产者 和 消费者 ——向 Kafka 写入数据 和读取数据
文章浏览阅读854次，点赞22次，收藏24次。点对点模型：适用于一对一的消息传递，具有高可靠性。发布/订阅模型：适用于广播消息给多个消费者，实现消息的广播。主题模型：适用于根据消息的主题进行灵活的过滤和匹配，处理复杂的消息路由需求。
kafka消息队列的初步探索
文章浏览阅读1.5k次，点赞2次，收藏3次。kafka 自动配置在KafkaAutoConfiguration
【可视化工具】Kafka-Offset Explorer使用
文章浏览阅读1.3w次，点赞6次，收藏33次。Offset Explorer（以前称为Kafka Tool）是一个用于管理和使Apache Kafka ®集群的GUI应用程序。它提供了一个直观的UI，允许人们快速查看Kafka集群中的对象以及存储在集群主题中的消息。它包含面向开发人员和管理员的功能。二、环境信息系统环境：windows 10版本：2.2Kafka版本：Kafka2.0.0三、安装和使用3.1 下载Offset Explorer 和安装下载到本地的 .exe文件Next安装路径 ，Next。_offset explorer
【云原生进阶之PaaS中间件】第三章Kafka-4.3.2-broker网络模型
文章浏览阅读1.3k次，点赞12次，收藏19次。kafka broker 在启动的时候，会根据你配置的listeners 初始化它的网络组件，用来接收外界的请求，这个listeners你可能没配置过，它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议，监听哪个端口，如果我们没有特殊的要求的话，使用它默认的配置就可以了，顶多是修改下端口这块。
Kafka: 详解、使用教程和示例
文章浏览阅读1.3k次，点赞2次，收藏2次。Kafka 是一个强大的分布式流处理平台，用于实时数据传输和处理。通过本文详细的介绍、使用教程和示例，你可以了解 Kafka 的核心概念、安装、创建 Topic、使用生产者和消费者，从而为构建现代分布式应用打下坚实的基础。无论是构建实时数据流平台、日志收集系统还是事件驱动架构，Kafka 都是一个可靠、高效的解决方案。_博客系统怎么使用kafka
【Spring从成神到升仙系列 四】从源码分析 Spring 事务的来龙去脉
文章浏览阅读3.5k次，点赞42次，收藏56次。对于Java开发者而言，关于 Spring ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。本期 Spring 源码解析系列文章，将带你领略 Spring 源码的奥秘。本期源码文章吸收了之前 Kafka 源码文章的错误，将不再一行一行的带大家分析源码，我们将一些不重要的分当做黑盒处理，以便我们更快、更有效的阅读源码。废话不多说，发车！
Kafka篇——Kafka消费者端常见配置，涵盖自动手动提交offset、poll消息细节、健康状态检查、新消费组消费offset规则以及指定分区等技术点配置，全面无死角，一篇文章拿下！
文章浏览阅读1.1k次，点赞14次，收藏16次。一、自动提交offset1、概念Kafka中默认是自动提交offset。消费者在poll到消息后默认情况下，会自动向Broker的_consumer_offsets主题提交当前主题-分区消费的偏移量2、自动提交offset和手动提交offset流程图3、在Java中实现配置4、自动提交offset问题自动提交会丢消息。因为如果消费者还没有消费完poll下来的消息就自动提交了偏移量，那么此时消费者挂了，于是下一个消费者会从已经提交的offset的下一个位置开始消费消息。_kafka中自动提交offsets
分布式 - 消息队列Kafka：Kafka生产者架构和配置参数
文章浏览阅读1.6k次。如果生产者发送消息的速度超过发送到服务器的速度，则会导致生产者空间不足，这个时候KafkaProducer的send（）方法调用要么被阻塞，要么抛出异常，这个取决于参数max.block.ms的配置，此参数的默认值为60000，即60秒。在默认情况下，生产者发送的消息是未经压缩的。如果应用程序调用send()方法的速度超过生产者将消息发送给服务器的速度，那么生产者的缓冲空间可能会被耗尽，后续的send()方法调用会等待内存空间被释放，如果在max.block.ms之后还没有可用空间，就抛出异常。_kafka producer 参数
Kafka快速入门(最新版3.6.0)
文章浏览阅读2.9k次，点赞3次，收藏10次。kafka解决通信问题_kafka3.6
nginx+rsyslog+kafka+clickhouse+grafana 实现nginx 网关监控
文章浏览阅读1.5k次，点赞9次，收藏11次。上面都配置完了之后可以先验证下，保证数据最终到ck，如果有问题，需要再每个节点调试，比如先调试nginx->rsyslog ，可以先不配置kafka 输出，配置为console或者文件输出都可以，具体这里就不写了。这里做了一个类型转换，因为nginx，request-time 单位是s，我想最终呈现在grafana 中是ms，所以这里做了转换，当然grafana中也可以做。kafka 相关部署这里不做赘述，只要创建一个topic 就可以。
Kafka中的enable-auto-commit和auto-commit-interval配置
文章浏览阅读1.4k次，点赞22次，收藏16次。Kafka中的enable-auto-commit和auto-commit-interval配置_auto-commit-interval
thingsboard规则链调用外部API——kafka
文章浏览阅读742次。thingsboard规则链调用外部 kafka_thingsboard kafka
Kafka 简介
文章浏览阅读1.3k次，点赞18次，收藏22次。Kafka_简介
从kafka如何保证数据一致性看通常数据一致性设计
文章浏览阅读1.1k次，点赞16次，收藏14次。在数据库系统中有个概念叫事务，事务的作用是为了保证数据的一致性，意思是要么数据成功，要么数据失败，不存在数据操作了一半的情况，这就是数据的一致性。在很多系统或者组件中，很多场景都需要保证数据的一致性，有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性，比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。
元数据管理-Atlas的介绍和使用(集成Hive、Solr、Kafka、Kerberos)
文章浏览阅读1.4k次。概述介绍架构发展架构原理类型系统介绍类型hive_table类型介绍DataSet类型定义Asset类型定义Referenceable类型定义Process类型定义Entities(实体)Attributes(属性)安装安装环境准备安装Solr-7.7.3安装Atlas2.1.0Atlas配置Atlas集成HbaseAtlas集成SolrAtlas集成KafkaAtlas Server配置Kerberos相关配置Atlas集成HiveAtlas启动Atlas使用Hive元数据初次导入Hive元数据增量同步。_atlas元数据管理
zookeeper + kafka
文章浏览阅读659次。Zookeeper是一个开源的分布式服务管理框架。存储业务服务节点元数据及状态信息，并负责通知再 ZooKeeper 上注册的服务几点状态给客户端。
Kafka-Kraft 模式架构部署
文章浏览阅读1.4k次。Kafka-Kraft 模式架构部署_kafka kraft部署
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot