devbox
天景科技苑
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SpringBoot整合Flink（施耐德PLC物联网信息采集）_springboot flink

作者：天景科技苑 | 2024-08-08 17:59:56

赞

踩

springboot flink

SpringBoot整合Flink（施耐德PLC物联网信息采集）

Linux环境安装kafka

前情：

施耐德PLC设备（TM200C16R）设置好信息采集程序，连接局域网，SpringBoot订阅MQTT主题，消息转至kafka，由flink接收并持久化到mysql数据库；

Wireshark抓包如下：

MQTTBox测试订阅如下：

已知参数：

服务器IP：139.220.193.14

端口号：1883

应用端账号：admin@tenlink

应用端密码：Tenlink@123

物联网账号：202303171001

物联网账号密码：03171001

订阅话题（topic）：

202303171001/p（发布话题，由设备发送，应用端接收）

202303171001/s（订阅话题，由应用端发送，设备接收）

订阅mqtt （前提是kafka是已经就绪状态且plc_thoroughfare主题是存在的）

  • maven pom

  1.         <dependency>
  2.             <groupId>org.eclipse.paho</groupId>
  3.             <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  4.             <version>1.2.5</version>
  5.         </dependency>

  • yaml配置

  1. spring:
  2.   kafka:
  3.     bootstrap-servers: ip:9092
  4.     producer:
  5.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  6.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  7.  
  8. ## 自定义
  9. kafka:
  10.   topics:
  11.     # kafka 主题
  12.     plc1: plc_thoroughfare
  13.  
  14. plc:
  15.   broker: tcp://139.220.193.14:1883
  16.   subscribe-topic:  202303171001/p
  17.   username: admin@tenlink
  18.   password: Tenlink@123
  19.   client-id: subscribe_client

  • 订阅mqtt并将报文发送到kafka主题

  1. import org.eclipse.paho.client.mqttv3.*;
  2. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.kafka.core.KafkaTemplate;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import javax.annotation.PostConstruct;
  11.  
  12. /**
  13.  * PLC 订阅消息
  14.  */
  15. @Component
  16. public class SubscribeSample {
  17.  
  18.     private static final Logger log = LoggerFactory.getLogger(SubscribeSample.class);
  19.  
  20.     @Autowired
  21.     private KafkaTemplate<String,Object> kafkaTemplate;
  22.  
  23.     @Value("${kafka.topics.plc1}")
  24.     private String plc1;
  25.     @Value("${plc.broker}")
  26.     private String broker;
  27.     @Value("${plc.subscribe-topic}")
  28.     private String subscribeTopic;
  29.     @Value("${plc.username}")
  30.     private String username;
  31.     @Value("${plc.password}")
  32.     private String password;
  33.     @Value("${plc.client-id}")
  34.     private String clientId;
  35.  
  36.     @PostConstruct
  37.     public void plcGather() {
  38.         int qos = 0;
  39.  
  40.         Thread thread = new Thread(new Runnable() {
  41.             @Override
  42.             public void run() {
  43.                 MqttClient client = null;
  44.                 try {
  45.                     client = new MqttClient(broker, clientId, new MemoryPersistence());
  46.                     // 连接参数
  47.                     MqttConnectOptions options = new MqttConnectOptions();
  48.                     options.setUserName(username);
  49.                     options.setPassword(password.toCharArray());
  50.                     options.setConnectionTimeout(60);
  51.                     options.setKeepAliveInterval(60);
  52.                     // 设置回调
  53.                     client.setCallback(new MqttCallback() {
  54.  
  55.                         public void connectionLost(Throwable cause) {
  56.                             System.out.println("connectionLost: " + cause.getMessage());
  57.                         }
  58.  
  59.                         public void messageArrived(String topic, MqttMessage message) {
  60.  
  61.                             String data = new String(message.getPayload());
  62.  
  63.                             kafkaTemplate.send(plc1,data).addCallback(success ->{
  64.                                 // 消息发送到的topic
  65.                                 String kafkaTopic = success.getRecordMetadata().topic();
  66.                                 // 消息发送到的分区
  67. //                                int partition = success.getRecordMetadata().partition();
  68.                                 // 消息在分区内的offset
  69. //                                long offset = success.getRecordMetadata().offset();
  70.                                 log.info("mqtt成功将消息:{},转入到kafka主题->{}", data,kafkaTopic);
  71.                             },failure ->{
  72.                                 throw new RuntimeException("发送消息失败:" + failure.getMessage());
  73.                             });
  74.                         }
  75.  
  76.                         public void deliveryComplete(IMqttDeliveryToken token) {
  77.                             log.info("deliveryComplete---------{}", token.isComplete());
  78.                         }
  79.  
  80.                     });
  81.                     client.connect(options);
  82.                     client.subscribe(subscribeTopic, qos);
  83.                 } catch (MqttException e) {
  84.                     e.printStackTrace();
  85.                 }
  86.             }
  87.         });
  88.  
  89.         thread.start();
  90.     }
  91. }

  • 采集报文测试（如下图表示成功，并且已经发送到了kafka主题上）

Flink接收kafka数据

  • maven pom

  1. <!--工具类 开始-->
  2.         <dependency>
  3.             <groupId>com.alibaba</groupId>
  4.             <artifactId>fastjson</artifactId>
  5.             <version>1.2.83</version>
  6.         </dependency>
  7.         <dependency>
  8.             <groupId>org.apache.commons</groupId>
  9.             <artifactId>commons-collections4</artifactId>
  10.             <version>4.4</version>
  11.         </dependency>
  12.         <dependency>
  13.             <groupId>org.apache.commons</groupId>
  14.             <artifactId>commons-lang3</artifactId>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>org.projectlombok</groupId>
  18.             <artifactId>lombok</artifactId>
  19.             <version>1.18.26</version>
  20.         </dependency>
  21.         <!--工具类 结束-->
  22.  
  23.         <!-- flink依赖引入 开始-->
  24.         <dependency>
  25.             <groupId>org.apache.flink</groupId>
  26.             <artifactId>flink-java</artifactId>
  27.             <version>1.13.1</version>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>org.apache.flink</groupId>
  31.             <artifactId>flink-streaming-java_2.11</artifactId>
  32.             <version>1.13.1</version>
  33.         </dependency>
  34.  
  35.         <dependency>
  36.             <groupId>org.apache.flink</groupId>
  37.             <artifactId>flink-clients_2.11</artifactId>
  38.             <version>1.13.1</version>
  39.         </dependency>
  40.         <!-- flink连接kafka -->
  41.         <dependency>
  42.             <groupId>org.apache.flink</groupId>
  43.             <artifactId>flink-connector-kafka_2.11</artifactId>
  44.             <version>1.13.1</version>
  45.         </dependency>
  46.         <!-- flink连接es-->
  47.         <dependency>
  48.             <groupId>org.apache.flink</groupId>
  49.             <artifactId>flink-json</artifactId>
  50.             <version>1.13.1</version>
  51.         </dependency>
  52.         <!-- flink连接mysql-->
  53.         <dependency>
  54.             <groupId>org.apache.flink</groupId>
  55.             <artifactId>flink-jdbc_2.11</artifactId>
  56.             <version>1.10.0</version>
  57.         </dependency>
  58.         <!-- flink依赖引入 结束-->
  59.  
  60.         <!--spring data jpa-->
  61.         <dependency>
  62.             <groupId>org.springframework.boot</groupId>
  63.             <artifactId>spring-boot-starter-data-jpa</artifactId>
  64.         </dependency>

  • yaml配置

  1. # 服务接口
  2. server:
  3.   port: 8222
  4.  
  5. spring:
  6.   kafka:
  7.     bootstrap-servers: ip:9092
  8.     consumer:
  9.       group-id: kafka
  10.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  11.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  12.  
  13.   datasource:
  14.     url:  jdbc:mysql://127.0.0.01:3306/ceshi?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai
  15.     driver-class-name:  com.mysql.cj.jdbc.Driver
  16.     username: root
  17.     password: root
  18.     druid:
  19.       initial-size: 5 #初始化时建立物理连接的个数
  20.       min-idle: 5 #最小连接池数量
  21.       maxActive: 20 #最大连接池数量
  22.       maxWait: 60000 #获取连接时最大等待时间，单位毫秒
  23.       timeBetweenEvictionRunsMillis: 60000 #配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
  24.       minEvictableIdleTimeMillis: 300000 #配置一个连接在池中最小生存的时间，单位是毫秒
  25.       validationQuery: SELECT 1 #用来检测连接是否有效的sql
  26.       testWhileIdle: true #申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效
  27.       testOnBorrow: false #申请连接时执行validationQuery检测连接是否有效，如果为true会降低性能
  28.       testOnReturn: false #归还连接时执行validationQuery检测连接是否有效，如果为true会降低性能
  29.       poolPreparedStatements: true # 打开PSCache，并且指定每个连接上PSCache的大小
  30.       maxPoolPreparedStatementPerConnectionSize: 20 #要启用PSCache，必须配置大于0，当大于0时，poolPreparedStatements自动触发修改为true。在Druid中，不会存在Oracle下PSCache占用内存过多的问题，可以把这个数值配置大一些，比如说100
  31.       filters: stat,wall,slf4j #配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
  32.       #通过connectProperties属性来打开mergeSql功能；慢SQL记录
  33.       connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
  34.  
  35.   jpa:
  36.     hibernate:
  37.       ddl-auto: none
  38.     show-sql: true
  39.     repositories:
  40.       packages: com.hzh.demo.domain.*
  41.  
  42. #自定义配置
  43. customer:
  44.   #flink相关配置
  45.   flink:
  46.     # 功能开关
  47.     plc-status: true
  48.     plc-topic: plc_thoroughfare
  49.  
  50. # 定时任务定时清理失效数据
  51. task:
  52.   plc-time: 0 0/1 * * * ?

  • 表结构

  1. -- plc_test definition
  2. CREATE TABLE `plc_test` (
  3.                             `pkid` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '主键id',
  4.                             `json_str` text CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT 'json格式数据',
  5.                             `create_time` bigint NOT NULL COMMENT '创建时间',
  6.                             PRIMARY KEY (`pkid`)
  7. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='plc存储数据测试表';

  • 启动类

  1. import org.springframework.boot.SpringApplication;
  2. import org.springframework.boot.autoconfigure.SpringBootApplication;
  3. import org.springframework.boot.autoconfigure.domain.EntityScan;
  4. import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
  5. import org.springframework.scheduling.annotation.EnableScheduling;
  6.  
  7. @SpringBootApplication
  8. @EnableJpaRepositories(basePackages = "repository basePackages")
  9. @EntityScan("entity basePackages")
  10. @EnableScheduling
  11. public class PLCStorageApplication {
  12.  
  13.     public static void main(String[] args) {
  14.         SpringApplication.run(PLCStorageApplication.class, args);
  15.     }
  16. }

  • 实体类

  1. import lombok.Builder;
  2. import lombok.Data;
  3.  
  4. import javax.persistence.Column;
  5. import javax.persistence.Entity;
  6. import javax.persistence.Id;
  7. import javax.persistence.Table;
  8. import java.io.Serializable;
  9.  
  10. /**
  11.  * PLC接收实体
  12.  */
  13. @Table(name = "plc_test")
  14. @Data
  15. @Builder
  16. @Entity
  17. public class PLCDomain implements Serializable {
  18.  
  19.     private static final long serialVersionUID = 4122384962907036649L;
  20.  
  21.     @Id
  22.     @Column(name = "pkid")
  23.     public String id;
  24.     @Column(name = "json_str")
  25.     public String jsonStr;
  26.     @Column(name = "create_time")
  27.     private Long createTime;
  28.  
  29.     public PLCDomain(String id, String jsonStr,Long createTime) {
  30.         this.id = id;
  31.         this.jsonStr = jsonStr;
  32.         this.createTime = createTime;
  33.     }
  34.  
  35.     public PLCDomain() {
  36.  
  37.     }
  38. }

  • jpa 接口 

  1. import com.hzh.demo.domain.PLCDomain;
  2. import org.springframework.data.jpa.repository.JpaRepository;
  3. import org.springframework.stereotype.Repository;
  4.  
  5. @Repository
  6. public interface PLCRepository extends JpaRepository<PLCDomain,String> {
  7.  
  8. }

  • 封装获取上下文工具类（ApplicationContextAware）由于加载先后顺序，flink无法使用spring bean注入的方式，特此封装工具类

  1. import org.springframework.beans.BeansException;
  2. import org.springframework.context.ApplicationContext;
  3. import org.springframework.context.ApplicationContextAware;
  4. import org.springframework.context.i18n.LocaleContextHolder;
  5. import org.springframework.stereotype.Component;
  6.  
  7.  
  8. @Component
  9. public class ApplicationContextProvider
  10.         implements ApplicationContextAware {
  11.     /**
  12.      * 上下文对象实例
  13.      */
  14.     private static ApplicationContext applicationContext;
  15.  
  16.     /**
  17.      * 获取applicationContext
  18.      *
  19.      * @return
  20.      */
  21.     public static ApplicationContext getApplicationContext() {
  22.         return applicationContext;
  23.     }
  24.  
  25.     @Override
  26.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  27.         ApplicationContextProvider.applicationContext = applicationContext;
  28.     }
  29.  
  30.     /**
  31.      * 通过name获取 Bean.
  32.      *
  33.      * @param name
  34.      * @return
  35.      */
  36.     public static Object getBean(String name) {
  37.         return getApplicationContext().getBean(name);
  38.     }
  39.  
  40.     /**
  41.      * 通过class获取Bean.
  42.      *
  43.      * @param clazz
  44.      * @param <T>
  45.      * @return
  46.      */
  47.     public static <T> T getBean(Class<T> clazz) {
  48.         return getApplicationContext().getBean(clazz);
  49.     }
  50.  
  51.     /**
  52.      * 通过name,以及Clazz返回指定的Bean
  53.      *
  54.      * @param name
  55.      * @param clazz
  56.      * @param <T>
  57.      * @return
  58.      */
  59.     public static <T> T getBean(String name, Class<T> clazz) {
  60.         return getApplicationContext().getBean(name, clazz);
  61.     }
  62.  
  63.  
  64.     /**
  65.      * 描述 : <获得多语言的资源内容>. <br>
  66.      * <p>
  67.      * <使用方法说明>
  68.      * </p>
  69.      *
  70.      * @param code
  71.      * @param args
  72.      * @return
  73.      */
  74.     public static String getMessage(String code, Object[] args) {
  75.         return getApplicationContext().getMessage(code, args, LocaleContextHolder.getLocale());
  76.     }
  77.  
  78.     /**
  79.      * 描述 : <获得多语言的资源内容>. <br>
  80.      * <p>
  81.      * <使用方法说明>
  82.      * </p>
  83.      *
  84.      * @param code
  85.      * @param args
  86.      * @param defaultMessage
  87.      * @return
  88.      */
  89.     public static String getMessage(String code, Object[] args,
  90.                                     String defaultMessage) {
  91.         return getApplicationContext().getMessage(code, args, defaultMessage,
  92.                 LocaleContextHolder.getLocale());
  93.     }
  94. }

  • FIink 第三方输出（mysql写入）

  1. import com.hzh.demo.config.ApplicationContextProvider;
  2. import com.hzh.demo.domain.PLCDomain;
  3. import com.hzh.demo.repository.PLCRepository;
  4. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import java.util.UUID;
  11.  
  12.  
  13. /**
  14.  * 向mysql写入数据
  15.  */
  16. @Component
  17. @ConditionalOnProperty(name = "customer.flink.plc-status")
  18. public class MysqlSink implements SinkFunction<String> {
  19.  
  20.     private static final Logger log = LoggerFactory.getLogger(MysqlSink.class);
  21.  
  22.     @Override
  23.     public void invoke(String value, Context context) throws Exception {
  24.         long currentTime = context.currentProcessingTime();
  25.         PLCDomain build = PLCDomain.builder()
  26.                 .id(UUID.randomUUID().toString().replaceAll("-", ""))
  27.                 .jsonStr(value)
  28.                 .createTime(currentTime)
  29.                 .build();
  30.  
  31.         PLCRepository repository = ApplicationContextProvider.getBean(PLCRepository.class);
  32.         repository.save(build);
  33.         log.info("持久化写入：{}",build);
  34.         SinkFunction.super.invoke(value, context);
  35.     }
  36. }

  • Flink订阅kafka topic读取持续数据

  1. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  2. import org.apache.flink.api.java.functions.KeySelector;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8. import org.springframework.beans.factory.annotation.Value;
  9. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import javax.annotation.PostConstruct;
  13. import java.util.Properties;
  14.  
  15. /**
  16.  * 接收 kafka topic 读取数据
  17.  */
  18. @Component
  19. @ConditionalOnProperty(name = "customer.flink.plc-status")
  20. public class FlinkReceivingPLC {
  21.     private static final Logger log = LoggerFactory.getLogger(MyKeyedProcessFunction.class);
  22.     @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
  23.     private String kafkaServer;
  24.     @Value("${customer.flink.plc-topic}")
  25.     private String topic;
  26.     @Value("${spring.kafka.consumer.group-id:kafka}")
  27.     private String groupId;
  28.     @Value("${spring.kafka.consumer.key-deserializer:org.apache.kafka.common.serialization.StringDeserializer}")
  29.     private String keyDeserializer;
  30.     @Value("${spring.kafka.consumer.value-deserializer:org.apache.kafka.common.serialization.StringDeserializer}")
  31.     private String valueDeserializer;
  32.  
  33.     /**
  34.      * 执行方法
  35.      *
  36.      * @throws Exception 异常
  37.      */
  38.     @PostConstruct
  39.     public void execute(){
  40.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  41.         env.enableCheckpointing(5000);
  42.         //设定全局并发度
  43.         env.setParallelism(1);
  44.         Properties properties = new Properties();
  45.         //kafka的节点的IP或者hostName，多个使用逗号分隔
  46.         properties.setProperty("bootstrap.servers", kafkaServer);
  47.         //kafka的消费者的group.id
  48.         properties.setProperty("group.id", groupId);
  49.         properties.setProperty("key-deserializer",keyDeserializer);
  50.         properties.setProperty("value-deserializer",valueDeserializer);
  51.  
  52.         FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), properties);
  53.  
  54.         DataStream<String> stream = env.addSource(myConsumer);
  55.         stream.print().setParallelism(1);
  56.  
  57.         stream
  58.                 //分组
  59.                 .keyBy(new KeySelector<String, String>() {
  60.                     @Override
  61.                     public String getKey(String value) throws Exception {
  62.                         return value;
  63.                     }
  64.                 })
  65.                 //指定处理类
  66. //                .process(new MyKeyedProcessFunction())
  67.                 //数据第三方输出，mysql持久化
  68.                 .addSink(new MysqlSink());
  69.  
  70.         //启动任务
  71.         new Thread(() -> {
  72.             try {
  73.                 env.execute("PLCPersistenceJob");
  74.             } catch (Exception e) {
  75.                 log.error(e.toString(), e);
  76.             }
  77.         }).start();
  78.     }
  79. }

  • 失效数据清理机制（为了方便测试，所以清理机制执行频率高且数据失效低）

  1. import com.hzh.demo.repository.PLCRepository;
  2. import org.slf4j.Logger;
  3. import org.slf4j.LoggerFactory;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.scheduling.annotation.Scheduled;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import java.util.Optional;
  10.  
  11. /**
  12.  * 定时任务配置
  13.  */
  14. @Component
  15. @Configuration
  16. public class QutrzConfig {
  17.  
  18.     private static final Logger log = LoggerFactory.getLogger(QutrzConfig.class);
  19.  
  20.     @Autowired
  21.     private PLCRepository plcRepository;
  22.  
  23.     /**
  24.      * 数据清理机制
  25.      */
  26.     @Scheduled(cron = "${task.plc-time}")
  27.     private void PLCCleaningMechanism (){
  28.  
  29.         log.info("执行数据清理机制：{}","PLCCleaningMechanism");
  30.  
  31.         long currentTimeMillis = System.currentTimeMillis();
  32.         Optional.of(this.plcRepository.findAll()).ifPresent(list ->{
  33.             list.forEach(plc ->{
  34.                 Long createTime = plc.getCreateTime();
  35.  
  36.                 //大于1分钟为失效数据
  37.                 if ((currentTimeMillis - createTime) > (1000 * 60 * 1) ){
  38.                     this.plcRepository.delete(plc);
  39.                     log.info("过期数据已经被清理：{}",plc);
  40.                 }
  41.             });
  42.         });
  43.     }
  44. }

  • 测试结果

  • mysql入库数据

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/天景科技苑/article/detail/949321
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号