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
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>yaml配置
spring:
kafka:
bootstrap-servers: ip:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
## 自定义
kafka:
topics:
# kafka 主题
plc1: plc_thoroughfare
plc:
broker: tcp://139.220.193.14:1883
subscribe-topic: 202303171001/p
username: admin@tenlink
password: Tenlink@123
client-id: subscribe_client订阅mqtt并将报文发送到kafka主题
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* PLC 订阅消息
*/
@Component
public class SubscribeSample {
private static final Logger log = LoggerFactory.getLogger(SubscribeSample.class);
@Autowired
private KafkaTemplate<String,Object> kafkaTemplate;
@Value("${kafka.topics.plc1}")
private String plc1;
@Value("${plc.broker}")
private String broker;
@Value("${plc.subscribe-topic}")
private String subscribeTopic;
@Value("${plc.username}")
private String username;
@Value("${plc.password}")
private String password;
@Value("${plc.client-id}")
private String clientId;
@PostConstruct
public void plcGather() {
int qos = 0;
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
MqttClient client = null;
try {
client = new MqttClient(broker, clientId, new MemoryPersistence());
// 连接参数
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
// 设置回调
client.setCallback(new MqttCallback() {
public void connectionLost(Throwable cause) {
System.out.println("connectionLost: " + cause.getMessage());
}
public void messageArrived(String topic, MqttMessage message) {
String data = new String(message.getPayload());
kafkaTemplate.send(plc1,data).addCallback(success ->{
// 消息发送到的topic
String kafkaTopic = success.getRecordMetadata().topic();
// 消息发送到的分区
// int partition = success.getRecordMetadata().partition();
// 消息在分区内的offset
// long offset = success.getRecordMetadata().offset();
log.info("mqtt成功将消息:{},转入到kafka主题->{}", data,kafkaTopic);
},failure ->{
throw new RuntimeException("发送消息失败:" + failure.getMessage());
});
}
public void deliveryComplete(IMqttDeliveryToken token) {
log.info("deliveryComplete---------{}", token.isComplete());
}
});
client.connect(options);
client.subscribe(subscribeTopic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
});
thread.start();
}
}
采集报文测试（如下图表示成功，并且已经发送到了kafka主题上）
Flink接收kafka数据
maven pom
<!--工具类 开始-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
<version>4.4</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.26</version>
</dependency>
<!--工具类 结束-->
<!-- flink依赖引入 开始-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.13.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.13.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.13.1</version>
</dependency>
<!-- flink连接kafka -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.13.1</version>
</dependency>
<!-- flink连接es-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>1.13.1</version>
</dependency>
<!-- flink连接mysql-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-jdbc_2.11</artifactId>
<version>1.10.0</version>
</dependency>
<!-- flink依赖引入 结束-->
<!--spring data jpa-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>yaml配置
# 服务接口
server:
port: 8222
spring:
kafka:
bootstrap-servers: ip:9092
consumer:
group-id: kafka
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
datasource:
url: jdbc:mysql://127.0.0.01:3306/ceshi?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai
driver-class-name: com.mysql.cj.jdbc.Driver
username: root
password: root
druid:
initial-size: 5 #初始化时建立物理连接的个数
min-idle: 5 #最小连接池数量
maxActive: 20 #最大连接池数量
maxWait: 60000 #获取连接时最大等待时间，单位毫秒
timeBetweenEvictionRunsMillis: 60000 #配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
minEvictableIdleTimeMillis: 300000 #配置一个连接在池中最小生存的时间，单位是毫秒
validationQuery: SELECT 1 #用来检测连接是否有效的sql
testWhileIdle: true #申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效
testOnBorrow: false #申请连接时执行validationQuery检测连接是否有效，如果为true会降低性能
testOnReturn: false #归还连接时执行validationQuery检测连接是否有效，如果为true会降低性能
poolPreparedStatements: true # 打开PSCache，并且指定每个连接上PSCache的大小
maxPoolPreparedStatementPerConnectionSize: 20 #要启用PSCache，必须配置大于0，当大于0时，poolPreparedStatements自动触发修改为true。在Druid中，不会存在Oracle下PSCache占用内存过多的问题，可以把这个数值配置大一些，比如说100
filters: stat,wall,slf4j #配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
#通过connectProperties属性来打开mergeSql功能；慢SQL记录
connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
jpa:
hibernate:
ddl-auto: none
show-sql: true
repositories:
packages: com.hzh.demo.domain.*
#自定义配置
customer:
#flink相关配置
flink:
# 功能开关
plc-status: true
plc-topic: plc_thoroughfare
# 定时任务定时清理失效数据
task:
plc-time: 0 0/1 * * * ?表结构
-- plc_test definition
CREATE TABLE `plc_test` (
`pkid` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '主键id',
`json_str` text CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT 'json格式数据',
`create_time` bigint NOT NULL COMMENT '创建时间',
PRIMARY KEY (`pkid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='plc存储数据测试表';启动类
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.domain.EntityScan;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication
@EnableJpaRepositories(basePackages = "repository basePackages")
@EntityScan("entity basePackages")
@EnableScheduling
public class PLCStorageApplication {
public static void main(String[] args) {
SpringApplication.run(PLCStorageApplication.class, args);
}
}实体类
import lombok.Builder;
import lombok.Data;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.Table;
import java.io.Serializable;
/**
* PLC接收实体
*/
@Table(name = "plc_test")
@Data
@Builder
@Entity
public class PLCDomain implements Serializable {
private static final long serialVersionUID = 4122384962907036649L;
@Id
@Column(name = "pkid")
public String id;
@Column(name = "json_str")
public String jsonStr;
@Column(name = "create_time")
private Long createTime;
public PLCDomain(String id, String jsonStr,Long createTime) {
this.id = id;
this.jsonStr = jsonStr;
this.createTime = createTime;
}
public PLCDomain() {
}
}
jpa 接口
import com.hzh.demo.domain.PLCDomain;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;
@Repository
public interface PLCRepository extends JpaRepository<PLCDomain,String> {
}封装获取上下文工具类（ApplicationContextAware）由于加载先后顺序，flink无法使用spring bean注入的方式，特此封装工具类
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.i18n.LocaleContextHolder;
import org.springframework.stereotype.Component;
@Component
public class ApplicationContextProvider
implements ApplicationContextAware {
/**
* 上下文对象实例
*/
private static ApplicationContext applicationContext;
/**
* 获取applicationContext
*
* @return
*/
public static ApplicationContext getApplicationContext() {
return applicationContext;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
ApplicationContextProvider.applicationContext = applicationContext;
}
/**
* 通过name获取 Bean.
*
* @param name
* @return
*/
public static Object getBean(String name) {
return getApplicationContext().getBean(name);
}
/**
* 通过class获取Bean.
*
* @param clazz
* @param <T>
* @return
*/
public static <T> T getBean(Class<T> clazz) {
return getApplicationContext().getBean(clazz);
}
/**
* 通过name,以及Clazz返回指定的Bean
*
* @param name
* @param clazz
* @param <T>
* @return
*/
public static <T> T getBean(String name, Class<T> clazz) {
return getApplicationContext().getBean(name, clazz);
}
/**
* 描述 : <获得多语言的资源内容>. <br>
* <p>
* <使用方法说明>
* </p>
*
* @param code
* @param args
* @return
*/
public static String getMessage(String code, Object[] args) {
return getApplicationContext().getMessage(code, args, LocaleContextHolder.getLocale());
}
/**
* 描述 : <获得多语言的资源内容>. <br>
* <p>
* <使用方法说明>
* </p>
*
* @param code
* @param args
* @param defaultMessage
* @return
*/
public static String getMessage(String code, Object[] args,
String defaultMessage) {
return getApplicationContext().getMessage(code, args, defaultMessage,
LocaleContextHolder.getLocale());
}
}
FIink 第三方输出（mysql写入）
import com.hzh.demo.config.ApplicationContextProvider;
import com.hzh.demo.domain.PLCDomain;
import com.hzh.demo.repository.PLCRepository;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.UUID;
/**
* 向mysql写入数据
*/
@Component
@ConditionalOnProperty(name = "customer.flink.plc-status")
public class MysqlSink implements SinkFunction<String> {
private static final Logger log = LoggerFactory.getLogger(MysqlSink.class);
@Override
public void invoke(String value, Context context) throws Exception {
long currentTime = context.currentProcessingTime();
PLCDomain build = PLCDomain.builder()
.id(UUID.randomUUID().toString().replaceAll("-", ""))
.jsonStr(value)
.createTime(currentTime)
.build();
PLCRepository repository = ApplicationContextProvider.getBean(PLCRepository.class);
repository.save(build);
log.info("持久化写入：{}",build);
SinkFunction.super.invoke(value, context);
}
}
Flink订阅kafka topic读取持续数据
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Properties;
/**
* 接收 kafka topic 读取数据
*/
@Component
@ConditionalOnProperty(name = "customer.flink.plc-status")
public class FlinkReceivingPLC {
private static final Logger log = LoggerFactory.getLogger(MyKeyedProcessFunction.class);
@Value("${spring.kafka.bootstrap-servers:localhost:9092}")
private String kafkaServer;
@Value("${customer.flink.plc-topic}")
private String topic;
@Value("${spring.kafka.consumer.group-id:kafka}")
private String groupId;
@Value("${spring.kafka.consumer.key-deserializer:org.apache.kafka.common.serialization.StringDeserializer}")
private String keyDeserializer;
@Value("${spring.kafka.consumer.value-deserializer:org.apache.kafka.common.serialization.StringDeserializer}")
private String valueDeserializer;
/**
* 执行方法
*
* @throws Exception 异常
*/
@PostConstruct
public void execute(){
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000);
//设定全局并发度
env.setParallelism(1);
Properties properties = new Properties();
//kafka的节点的IP或者hostName，多个使用逗号分隔
properties.setProperty("bootstrap.servers", kafkaServer);
//kafka的消费者的group.id
properties.setProperty("group.id", groupId);
properties.setProperty("key-deserializer",keyDeserializer);
properties.setProperty("value-deserializer",valueDeserializer);
FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), properties);
DataStream<String> stream = env.addSource(myConsumer);
stream.print().setParallelism(1);
stream
//分组
.keyBy(new KeySelector<String, String>() {
@Override
public String getKey(String value) throws Exception {
return value;
}
})
//指定处理类
// .process(new MyKeyedProcessFunction())
//数据第三方输出，mysql持久化
.addSink(new MysqlSink());
//启动任务
new Thread(() -> {
try {
env.execute("PLCPersistenceJob");
} catch (Exception e) {
log.error(e.toString(), e);
}
}).start();
}
}失效数据清理机制（为了方便测试，所以清理机制执行频率高且数据失效低）
import com.hzh.demo.repository.PLCRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
* 定时任务配置
*/
@Component
@Configuration
public class QutrzConfig {
private static final Logger log = LoggerFactory.getLogger(QutrzConfig.class);
@Autowired
private PLCRepository plcRepository;
/**
* 数据清理机制
*/
@Scheduled(cron = "${task.plc-time}")
private void PLCCleaningMechanism (){
log.info("执行数据清理机制：{}","PLCCleaningMechanism");
long currentTimeMillis = System.currentTimeMillis();
Optional.of(this.plcRepository.findAll()).ifPresent(list ->{
list.forEach(plc ->{
Long createTime = plc.getCreateTime();
//大于1分钟为失效数据
if ((currentTimeMillis - createTime) > (1000 * 60 * 1) ){
this.plcRepository.delete(plc);
log.info("过期数据已经被清理：{}",plc);
}
});
});
}
}
测试结果
mysql入库数据
