Flink+Kafka+Mysql(8.0)

一、前言阐明

Flink版本：1.14.6
Mysql版本：8.0
JDK版本：1.8
本实例主要实现功能如下：
模拟消息生成->Kafka->Flink->Mysql
其中Flink做数据流网络并定时批量写入到Mysql

本例使用Intellij IDEA作为项目开发的IDE。
整个项目结构如图所示：

二、项目代码已提交至 gitee

flink-kafka-mysql: Flink读取Kafka 消息并批量写入到 MySQL8.0
三、项目具体代码

POM文件内容：

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <groupId>org.example</groupId>
8.     <artifactId>flink-kafka-mysql</artifactId>
9.     <version>1.0-SNAPSHOT</version>
11.     <properties>
12.         <maven.compiler.source>8</maven.compiler.source>
13.         <maven.compiler.target>8</maven.compiler.target>
14.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15.         <flink.version>1.14.6</flink.version>
16.         <scala.binary.version>2.11</scala.binary.version>
17.     </properties>
19.     <dependencies>
20.         <dependency>
21.             <groupId>org.apache.flink</groupId>
22.             <artifactId>flink-java</artifactId>
23.             <version>${flink.version}</version>
24.         </dependency>
25.         <dependency>
26.             <groupId>org.apache.flink</groupId>
27.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
28.             <version>${flink.version}</version>
29.         </dependency>
30.         <dependency>
31.             <groupId>org.apache.flink</groupId>
32.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
33.             <version>${flink.version}</version>
34.         </dependency>
35.         <dependency>
36.             <groupId>org.apache.flink</groupId>
37.             <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
38.             <version>${flink.version}</version>
39.         </dependency>
40.         <dependency>
41.             <groupId>org.apache.flink</groupId>
42.             <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
43.             <version>${flink.version}</version>
44.         </dependency>
45.         <dependency>
46.             <groupId>org.slf4j</groupId>
47.             <artifactId>slf4j-log4j12</artifactId>
48.             <version>1.7.25</version>
49.         </dependency>
50.         <dependency>
51.             <groupId>log4j</groupId>
52.             <artifactId>log4j</artifactId>
53.             <version>1.2.17</version>
54.         </dependency>
55.         <dependency>
56.             <groupId>mysql</groupId>
57.             <artifactId>mysql-connector-java</artifactId>
58.             <version>8.0.23</version>
59.         </dependency>
60.         <dependency>
61.             <groupId>com.alibaba</groupId>
62.             <artifactId>druid</artifactId>
63.             <version>1.2.8</version>
64.         </dependency>
65.         <dependency>
66.             <groupId>org.apache.kafka</groupId>
67.             <artifactId>kafka-clients</artifactId>
68.             <version>2.1.1</version>
69.         </dependency>
70.         <dependency>
71.             <groupId>com.alibaba</groupId>
72.             <artifactId>fastjson</artifactId>
73.             <version>1.2.70</version>
74.         </dependency>
75.     </dependencies>
78.     <build>
79.         <plugins>
80.             <!-- 构建主 JAR 文件（瘦包） -->
81.             <plugin>
82.                 <groupId>org.apache.maven.plugins</groupId>
83.                 <artifactId>maven-jar-plugin</artifactId>
84.                 <version>3.2.0</version>
85.                 <configuration>
86.                     <archive>
87.                         <manifest>
88.                             <addClasspath>true</addClasspath>
89.                             <classpathPrefix>lib/</classpathPrefix> <!-- 设置依赖的类路径前缀为 lib/ -->
90.                             <mainClass>com.slink.StudentRunner</mainClass> <!-- 替换为你的主类 -->
91.                         </manifest>
92.                     </archive>
93.                 </configuration>
94.             </plugin>
96.             <!-- 将依赖项复制到 lib 目录 -->
97.             <plugin>
98.                 <groupId>org.apache.maven.plugins</groupId>
99.                 <artifactId>maven-dependency-plugin</artifactId>
100.                 <version>3.1.2</version>
101.                 <executions>
102.                     <execution>
103.                         <id>copy-dependencies</id>
104.                         <phase>package</phase>
105.                         <goals>
106.                             <goal>copy-dependencies</goal>
107.                         </goals>
108.                         <configuration>
109.                             <!-- 输出目录为 target/lib -->
110.                             <outputDirectory>${project.build.directory}/lib</outputDirectory>
111.                             <!-- 指定复制运行时依赖 -->
112.                             <includeScope>runtime</includeScope>
113.                         </configuration>
114.                     </execution>
115.                 </executions>
116.             </plugin>
118.             <!-- 可选：使用 maven-assembly-plugin 打包主 JAR 和 lib 目录 -->
119.             <plugin>
120.                 <groupId>org.apache.maven.plugins</groupId>
121.                 <artifactId>maven-assembly-plugin</artifactId>
122.                 <version>3.3.0</version>
123.                 <executions>
124.                     <execution>
125.                         <id>make-assembly</id>
126.                         <phase>package</phase>
127.                         <goals>
128.                             <goal>single</goal>
129.                         </goals>
130.                         <configuration>
131.                             <descriptors>
132.                                 <descriptor>src/main/resources/assembly/assembly.xml</descriptor>
133.                             </descriptors>
134.                         </configuration>
135.                     </execution>
136.                 </executions>
137.             </plugin>
138.         </plugins>
139.     </build>
140. </project>

复制代码

application-dev.properties配置文件：

3. # mysql-jdbc
4. jdbc.driver=com.mysql.cj.jdbc.Driver
5. jdbc.url=jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8&rewriteBatchedStatements=true
6. jdbc.username=root
7. jdbc.password=root
8. jdbc.druidInitialSize=10
9. jdbc.druidMaxActive=20
10. jdbc.druidMinIdle=5
11. jdbc.druidMaxWait=20000
12. jdbc.druidTimeBetweenEvictionRunsMillis=60000
13. jdbc.druidMaxEvictableIdleTimeMillis=3600000
14. jdbc.druidMinEvictableIdleTimeMillis=3240000
15. jdbc.druidTestWhileIdle=true
16. jdbc.druidTestOnBorrow=true
17. jdbc.druidTestOnReturn=false
18. jdbc.druidPoolPreparedStatements=true
19. jdbc.druidMaxPoolPreparedStatementPerConnectionSize=10
20. jdbc.druidFilters=stat,slf4j
21. jdbc.druidValidationQuery=select 1
23. # kafka
24. kafka.servers=127.0.0.1:9092
25. kafka.consumer.groupId=kafka.consumer.group
26. kafka.auto.offset.reset=latest
27. kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
28. kafka.value.serializer=org.apache.kafka.common.serialization.StringSerializer
29. kafka.user.event.topic=testTopic

复制代码

Mysql数据表创建：

2. CREATE TABLE `student` (
3.                            `id` bigint(20) NOT NULL AUTO_INCREMENT,
4.                            `name1` varchar(50),
5.                            `name2` varchar(50),
6.                            `name3` varchar(50),
7.                            `name4` varchar(50),
8.                            `name5` varchar(50),
9.                            `name6` varchar(50),
10.                            `name7` varchar(50),
11.                            `name8` varchar(50),
12.                            `name9` varchar(50),
13.                            `name10` varchar(50),
14.                            `name11` varchar(50),
15.                            `name12` varchar(50),
16.                            `name13` varchar(50),
17.                            `name14` varchar(50),
18.                            `name15` varchar(50),
19.                            `name16` varchar(50),
20.                            `name17` varchar(50),
21.                            `name18` varchar(50),
22.                            `name19` varchar(50),
23.                            `name20` varchar(50),
24.                            `name21` varchar(50),
25.                            `name22` varchar(50),
26.                            `name23` varchar(50),
27.                            `name24` varchar(50),
28.                            `name25` varchar(50),
29.                            `name26` varchar(50),
30.                            `name27` varchar(50),
31.                            `name28` varchar(50),
32.                            `name29` varchar(50),
33.                            `name30` varchar(50),
34.                            `name31` varchar(50),
35.                            `name32` varchar(50),
36.                            `name33` varchar(50),
37.                            `name34` varchar(50),
38.                            `name35` varchar(50),
39.                            `name36` varchar(50),
40.                            `name37` varchar(50),
41.                            `name38` varchar(50),
42.                            `name39` varchar(50),
43.                            `name40` varchar(50),
44.                            `name41` varchar(50),
45.                            `name42` varchar(50),
46.                            `name43` varchar(50),
47.                            `name44` varchar(50),
48.                            `name45` varchar(50),
49.                            `name46` varchar(50),
50.                            `name47` varchar(50),
51.                            `name48` varchar(50),
52.                            `name49` varchar(50),
53.                            `name50` varchar(50),
54.                            PRIMARY KEY (`id`)
55. ) ENGINE=InnoDB AUTO_INCREMENT=0 DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='学生表';

复制代码

创建Student实体类：

1. package com.slink.entity;
3. import java.io.Serializable;
5. public class Student implements Serializable {
6.     private long id;
7.     private String name1;
8.     private String name2;
9.     private String name3;
10.     private String name4;
11.     private String name5;
12.     private String name6;
13.     private String name7;
14.     private String name8;
15.     private String name9;
16.     private String name10;
17.     private String name11;
18.     private String name12;
19.     private String name13;
20.     private String name14;
21.     private String name15;
22.     private String name16;
23.     private String name17;
24.     private String name18;
25.     private String name19;
26.     private String name20;
27.     private String name21;
28.     private String name22;
29.     private String name23;
30.     private String name24;
31.     private String name25;
32.     private String name26;
33.     private String name27;
34.     private String name28;
35.     private String name29;
36.     private String name30;
37.     private String name31;
38.     private String name32;
39.     private String name33;
40.     private String name34;
41.     private String name35;
42.     private String name36;
43.     private String name37;
44.     private String name38;
45.     private String name39;
46.     private String name40;
47.     private String name41;
48.     private String name42;
49.     private String name43;
50.     private String name44;
51.     private String name45;
52.     private String name46;
53.     private String name47;
54.     private String name48;
55.     private String name49;
56.     private String name50;
58.     public long getId() {
59.         return id;
60.     }
62.     public void setId(long id) {
63.         this.id = id;
64.     }
66.     public String getName1() {
67.         return name1;
68.     }
70.     public void setName1(String name1) {
71.         this.name1 = name1;
72.     }
74.     public String getName2() {
75.         return name2;
76.     }
78.     public void setName2(String name2) {
79.         this.name2 = name2;
80.     }
82.     public String getName3() {
83.         return name3;
84.     }
86.     public void setName3(String name3) {
87.         this.name3 = name3;
88.     }
90.     public String getName4() {
91.         return name4;
92.     }
94.     public void setName4(String name4) {
95.         this.name4 = name4;
96.     }
98.     public String getName5() {
99.         return name5;
100.     }
102.     public void setName5(String name5) {
103.         this.name5 = name5;
104.     }
106.     public String getName6() {
107.         return name6;
108.     }
110.     public void setName6(String name6) {
111.         this.name6 = name6;
112.     }
114.     public String getName7() {
115.         return name7;
116.     }
118.     public void setName7(String name7) {
119.         this.name7 = name7;
120.     }
122.     public String getName8() {
123.         return name8;
124.     }
126.     public void setName8(String name8) {
127.         this.name8 = name8;
128.     }
130.     public String getName9() {
131.         return name9;
132.     }
134.     public void setName9(String name9) {
135.         this.name9 = name9;
136.     }
138.     public String getName10() {
139.         return name10;
140.     }
142.     public void setName10(String name10) {
143.         this.name10 = name10;
144.     }
146.     public String getName11() {
147.         return name11;
148.     }
150.     public void setName11(String name11) {
151.         this.name11 = name11;
152.     }
154.     public String getName12() {
155.         return name12;
156.     }
158.     public void setName12(String name12) {
159.         this.name12 = name12;
160.     }
162.     public String getName13() {
163.         return name13;
164.     }
166.     public void setName13(String name13) {
167.         this.name13 = name13;
168.     }
170.     public String getName14() {
171.         return name14;
172.     }
174.     public void setName14(String name14) {
175.         this.name14 = name14;
176.     }
178.     public String getName15() {
179.         return name15;
180.     }
182.     public void setName15(String name15) {
183.         this.name15 = name15;
184.     }
186.     public String getName16() {
187.         return name16;
188.     }
190.     public void setName16(String name16) {
191.         this.name16 = name16;
192.     }
194.     public String getName17() {
195.         return name17;
196.     }
198.     public void setName17(String name17) {
199.         this.name17 = name17;
200.     }
202.     public String getName18() {
203.         return name18;
204.     }
206.     public void setName18(String name18) {
207.         this.name18 = name18;
208.     }
210.     public String getName19() {
211.         return name19;
212.     }
214.     public void setName19(String name19) {
215.         this.name19 = name19;
216.     }
218.     public String getName20() {
219.         return name20;
220.     }
222.     public void setName20(String name20) {
223.         this.name20 = name20;
224.     }
226.     public String getName21() {
227.         return name21;
228.     }
230.     public void setName21(String name21) {
231.         this.name21 = name21;
232.     }
234.     public String getName22() {
235.         return name22;
236.     }
238.     public void setName22(String name22) {
239.         this.name22 = name22;
240.     }
242.     public String getName23() {
243.         return name23;
244.     }
246.     public void setName23(String name23) {
247.         this.name23 = name23;
248.     }
250.     public String getName24() {
251.         return name24;
252.     }
254.     public void setName24(String name24) {
255.         this.name24 = name24;
256.     }
258.     public String getName25() {
259.         return name25;
260.     }
262.     public void setName25(String name25) {
263.         this.name25 = name25;
264.     }
266.     public String getName26() {
267.         return name26;
268.     }
270.     public void setName26(String name26) {
271.         this.name26 = name26;
272.     }
274.     public String getName27() {
275.         return name27;
276.     }
278.     public void setName27(String name27) {
279.         this.name27 = name27;
280.     }
282.     public String getName28() {
283.         return name28;
284.     }
286.     public void setName28(String name28) {
287.         this.name28 = name28;
288.     }
290.     public String getName29() {
291.         return name29;
292.     }
294.     public void setName29(String name29) {
295.         this.name29 = name29;
296.     }
298.     public String getName30() {
299.         return name30;
300.     }
302.     public void setName30(String name30) {
303.         this.name30 = name30;
304.     }
306.     public String getName31() {
307.         return name31;
308.     }
310.     public void setName31(String name31) {
311.         this.name31 = name31;
312.     }
314.     public String getName32() {
315.         return name32;
316.     }
318.     public void setName32(String name32) {
319.         this.name32 = name32;
320.     }
322.     public String getName33() {
323.         return name33;
324.     }
326.     public void setName33(String name33) {
327.         this.name33 = name33;
328.     }
330.     public String getName34() {
331.         return name34;
332.     }
334.     public void setName34(String name34) {
335.         this.name34 = name34;
336.     }
338.     public String getName35() {
339.         return name35;
340.     }
342.     public void setName35(String name35) {
343.         this.name35 = name35;
344.     }
346.     public String getName36() {
347.         return name36;
348.     }
350.     public void setName36(String name36) {
351.         this.name36 = name36;
352.     }
354.     public String getName37() {
355.         return name37;
356.     }
358.     public void setName37(String name37) {
359.         this.name37 = name37;
360.     }
362.     public String getName38() {
363.         return name38;
364.     }
366.     public void setName38(String name38) {
367.         this.name38 = name38;
368.     }
370.     public String getName39() {
371.         return name39;
372.     }
374.     public void setName39(String name39) {
375.         this.name39 = name39;
376.     }
378.     public String getName40() {
379.         return name40;
380.     }
382.     public void setName40(String name40) {
383.         this.name40 = name40;
384.     }
386.     public String getName41() {
387.         return name41;
388.     }
390.     public void setName41(String name41) {
391.         this.name41 = name41;
392.     }
394.     public String getName42() {
395.         return name42;
396.     }
398.     public void setName42(String name42) {
399.         this.name42 = name42;
400.     }
402.     public String getName43() {
403.         return name43;
404.     }
406.     public void setName43(String name43) {
407.         this.name43 = name43;
408.     }
410.     public String getName44() {
411.         return name44;
412.     }
414.     public void setName44(String name44) {
415.         this.name44 = name44;
416.     }
418.     public String getName45() {
419.         return name45;
420.     }
422.     public void setName45(String name45) {
423.         this.name45 = name45;
424.     }
426.     public String getName46() {
427.         return name46;
428.     }
430.     public void setName46(String name46) {
431.         this.name46 = name46;
432.     }
434.     public String getName47() {
435.         return name47;
436.     }
438.     public void setName47(String name47) {
439.         this.name47 = name47;
440.     }
442.     public String getName48() {
443.         return name48;
444.     }
446.     public void setName48(String name48) {
447.         this.name48 = name48;
448.     }
450.     public String getName49() {
451.         return name49;
452.     }
454.     public void setName49(String name49) {
455.         this.name49 = name49;
456.     }
458.     public String getName50() {
459.         return name50;
460.     }
462.     public void setName50(String name50) {
463.         this.name50 = name50;
464.     }
465. }

复制代码

创建吸取Kafka消息的实体SinkLog：

1. package com.slink.entity;
3. import com.alibaba.fastjson.JSON;
4. import com.slink.util.EmptyNullUtil;
5. import org.apache.commons.lang3.StringUtils;
6. import org.slf4j.Logger;
7. import org.slf4j.LoggerFactory;
9. import java.io.Serializable;
12. public class SinkLog implements Serializable {
14.     private static final Logger log = LoggerFactory.getLogger(SinkLog.class);
16.     /**
17.      * 请求时间(2024-10-28T17:06:51+08:00)
18.      */
19.     private String timestamp;
21.     /**
22.      * 客户端地址
23.      */
24.     private String remoteAddr;
26.     /**
27.      * HTTP请求状态
28.      */
29.     private int status;
31.     /**
32.      * 请求的URI和HTTP协议(POST /sink/ HTTP/1.1)
33.      */
34.     private String request;
36.     /**
37.      * 后台upstream的地址，即真正提供服务的主机地址(127.0.0.1:80)
38.      */
39.     private String upstrAddr;
41.     /**
42.      * 发送给客户端文件内容大小
43.      */
44.     private long bytes;
46.     /**
47.      * 请求参数主体
48.      */
49.     private Student requestBody;
51.     /**
52.      * 用户终端浏览器等信息(Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; SV1; GTB7.0; .NET4.0C)
53.      */
54.     private String agent;
56.     public String getTimestamp() {
57.         return timestamp;
58.     }
60.     public void setTimestamp(String timestamp) {
61.         this.timestamp = timestamp;
62.     }
64.     public String getRemoteAddr() {
65.         return remoteAddr;
66.     }
68.     public void setRemoteAddr(String remoteAddr) {
69.         this.remoteAddr = remoteAddr;
70.     }
72.     public int getStatus() {
73.         return status;
74.     }
76.     public void setStatus(int status) {
77.         this.status = status;
78.     }
80.     public String getRequest() {
81.         return request;
82.     }
84.     public void setRequest(String request) {
85.         this.request = request;
86.     }
88.     public String getUpstrAddr() {
89.         return upstrAddr;
90.     }
92.     public void setUpstrAddr(String upstrAddr) {
93.         this.upstrAddr = upstrAddr;
94.     }
96.     public long getBytes() {
97.         return bytes;
98.     }
100.     public void setBytes(long bytes) {
101.         this.bytes = bytes;
102.     }
104.     public Student getRequestBody() {
105.         return requestBody;
106.     }
108.     public void setRequestBody(Student requestBody) {
109.         this.requestBody = requestBody;
110.     }
112.     public String getAgent() {
113.         return agent;
114.     }
116.     public void setAgent(String agent) {
117.         this.agent = agent;
118.     }
120.     /**
121.      * 解析字符串成Student对象
122.      */
123.     public static Student build(String val) {
124.         if (StringUtils.isEmpty(val) || !val.contains("/sink")) {
125.             return null;
126.         }
127.         log.info("消息:{}", val);
128.         try {
129.             val = val.replace("\", StringUtils.EMPTY);
130.             Student student = JSON.parseObject(val, SinkLog.class).getRequestBody();
131.             EmptyNullUtil.stringNullToEmpty(student);
132.             return student;
133.         } catch (Exception e) {
134.             log.error("解析字符串成Student对象异常:", e);
135.             return null;
136.         }
137.     }
138. }

复制代码

创建读取配置文件实体（SinkProperties）以及方法：

1. package com.slink.properties;
3. import org.apache.commons.lang3.StringUtils;
5. import java.io.IOException;
6. import java.io.InputStream;
7. import java.io.Serializable;
8. import java.util.Properties;
10. public class SinkProperties implements Serializable {
12.     private String jdbcDriver;
13.     private String jdbcUrl;
14.     private String jdbcUsername;
15.     private String jdbcPassword;
16.     private int jdbcDruidInitialSize;
17.     private int jdbcDruidMaxActive;
18.     private int jdbcDruidMinIdle;
19.     private int jdbcDruidMaxWait;
20.     private long jdbcDruidTimeBetweenEvictionRunsMillis;
21.     private long jdbcDruidMaxEvictableIdleTimeMillis;
22.     private long jdbcDruidMinEvictableIdleTimeMillis;
23.     private boolean jdbcDruidTestWhileIdle;
24.     private boolean jdbcDruidTestOnBorrow;
25.     private boolean jdbcDruidTestOnReturn;
26.     private boolean jdbcDruidPoolPreparedStatements;
27.     private int jdbcDruidMaxPoolPreparedStatementPerConnectionSize;
28.     private String jdbcDruidFilters;
29.     private String jdbcDruidValidationQuery;
30.     private String kafkaServer;
31.     private String kafkaConsumerGroupId;
32.     private String kafkaKeySerializer;
33.     private String kafkaValueSerializer;
34.     private String kafkaUserEventTopic;
35.     private String kafkaAutoOffsetReset;
37.     public String getJdbcDriver() {
38.         return jdbcDriver;
39.     }
41.     public void setJdbcDriver(String jdbcDriver) {
42.         this.jdbcDriver = jdbcDriver;
43.     }
45.     public String getJdbcUrl() {
46.         return jdbcUrl;
47.     }
49.     public void setJdbcUrl(String jdbcUrl) {
50.         this.jdbcUrl = jdbcUrl;
51.     }
53.     public String getJdbcUsername() {
54.         return jdbcUsername;
55.     }
57.     public void setJdbcUsername(String jdbcUsername) {
58.         this.jdbcUsername = jdbcUsername;
59.     }
61.     public String getJdbcPassword() {
62.         return jdbcPassword;
63.     }
65.     public void setJdbcPassword(String jdbcPassword) {
66.         this.jdbcPassword = jdbcPassword;
67.     }
69.     public int getJdbcDruidInitialSize() {
70.         return jdbcDruidInitialSize;
71.     }
73.     public void setJdbcDruidInitialSize(int jdbcDruidInitialSize) {
74.         this.jdbcDruidInitialSize = jdbcDruidInitialSize;
75.     }
77.     public int getJdbcDruidMaxActive() {
78.         return jdbcDruidMaxActive;
79.     }
81.     public void setJdbcDruidMaxActive(int jdbcDruidMaxActive) {
82.         this.jdbcDruidMaxActive = jdbcDruidMaxActive;
83.     }
85.     public int getJdbcDruidMinIdle() {
86.         return jdbcDruidMinIdle;
87.     }
89.     public void setJdbcDruidMinIdle(int jdbcDruidMinIdle) {
90.         this.jdbcDruidMinIdle = jdbcDruidMinIdle;
91.     }
93.     public int getJdbcDruidMaxWait() {
94.         return jdbcDruidMaxWait;
95.     }
97.     public void setJdbcDruidMaxWait(int jdbcDruidMaxWait) {
98.         this.jdbcDruidMaxWait = jdbcDruidMaxWait;
99.     }
101.     public long getJdbcDruidTimeBetweenEvictionRunsMillis() {
102.         return jdbcDruidTimeBetweenEvictionRunsMillis;
103.     }
105.     public void setJdbcDruidTimeBetweenEvictionRunsMillis(long jdbcDruidTimeBetweenEvictionRunsMillis) {
106.         this.jdbcDruidTimeBetweenEvictionRunsMillis = jdbcDruidTimeBetweenEvictionRunsMillis;
107.     }
109.     public long getJdbcDruidMaxEvictableIdleTimeMillis() {
110.         return jdbcDruidMaxEvictableIdleTimeMillis;
111.     }
113.     public void setJdbcDruidMaxEvictableIdleTimeMillis(long jdbcDruidMaxEvictableIdleTimeMillis) {
114.         this.jdbcDruidMaxEvictableIdleTimeMillis = jdbcDruidMaxEvictableIdleTimeMillis;
115.     }
117.     public long getJdbcDruidMinEvictableIdleTimeMillis() {
118.         return jdbcDruidMinEvictableIdleTimeMillis;
119.     }
121.     public void setJdbcDruidMinEvictableIdleTimeMillis(long jdbcDruidMinEvictableIdleTimeMillis) {
122.         this.jdbcDruidMinEvictableIdleTimeMillis = jdbcDruidMinEvictableIdleTimeMillis;
123.     }
125.     public boolean isJdbcDruidTestWhileIdle() {
126.         return jdbcDruidTestWhileIdle;
127.     }
129.     public void setJdbcDruidTestWhileIdle(boolean jdbcDruidTestWhileIdle) {
130.         this.jdbcDruidTestWhileIdle = jdbcDruidTestWhileIdle;
131.     }
133.     public boolean isJdbcDruidTestOnBorrow() {
134.         return jdbcDruidTestOnBorrow;
135.     }
137.     public void setJdbcDruidTestOnBorrow(boolean jdbcDruidTestOnBorrow) {
138.         this.jdbcDruidTestOnBorrow = jdbcDruidTestOnBorrow;
139.     }
141.     public boolean isJdbcDruidTestOnReturn() {
142.         return jdbcDruidTestOnReturn;
143.     }
145.     public void setJdbcDruidTestOnReturn(boolean jdbcDruidTestOnReturn) {
146.         this.jdbcDruidTestOnReturn = jdbcDruidTestOnReturn;
147.     }
149.     public boolean isJdbcDruidPoolPreparedStatements() {
150.         return jdbcDruidPoolPreparedStatements;
151.     }
153.     public void setJdbcDruidPoolPreparedStatements(boolean jdbcDruidPoolPreparedStatements) {
154.         this.jdbcDruidPoolPreparedStatements = jdbcDruidPoolPreparedStatements;
155.     }
157.     public int getJdbcDruidMaxPoolPreparedStatementPerConnectionSize() {
158.         return jdbcDruidMaxPoolPreparedStatementPerConnectionSize;
159.     }
161.     public void setJdbcDruidMaxPoolPreparedStatementPerConnectionSize(int jdbcDruidMaxPoolPreparedStatementPerConnectionSize) {
162.         this.jdbcDruidMaxPoolPreparedStatementPerConnectionSize = jdbcDruidMaxPoolPreparedStatementPerConnectionSize;
163.     }
165.     public String getJdbcDruidFilters() {
166.         return jdbcDruidFilters;
167.     }
169.     public void setJdbcDruidFilters(String jdbcDruidFilters) {
170.         this.jdbcDruidFilters = jdbcDruidFilters;
171.     }
173.     public String getJdbcDruidValidationQuery() {
174.         return jdbcDruidValidationQuery;
175.     }
177.     public void setJdbcDruidValidationQuery(String jdbcDruidValidationQuery) {
178.         this.jdbcDruidValidationQuery = jdbcDruidValidationQuery;
179.     }
181.     public String getKafkaServer() {
182.         return kafkaServer;
183.     }
185.     public void setKafkaServer(String kafkaServer) {
186.         this.kafkaServer = kafkaServer;
187.     }
189.     public String getKafkaConsumerGroupId() {
190.         return kafkaConsumerGroupId;
191.     }
193.     public void setKafkaConsumerGroupId(String kafkaConsumerGroupId) {
194.         this.kafkaConsumerGroupId = kafkaConsumerGroupId;
195.     }
197.     public String getKafkaKeySerializer() {
198.         return kafkaKeySerializer;
199.     }
201.     public void setKafkaKeySerializer(String kafkaKeySerializer) {
202.         this.kafkaKeySerializer = kafkaKeySerializer;
203.     }
205.     public String getKafkaValueSerializer() {
206.         return kafkaValueSerializer;
207.     }
209.     public void setKafkaValueSerializer(String kafkaValueSerializer) {
210.         this.kafkaValueSerializer = kafkaValueSerializer;
211.     }
213.     public String getKafkaUserEventTopic() {
214.         return kafkaUserEventTopic;
215.     }
217.     public void setKafkaUserEventTopic(String kafkaUserEventTopic) {
218.         this.kafkaUserEventTopic = kafkaUserEventTopic;
219.     }
221.     public String getKafkaAutoOffsetReset() {
222.         return kafkaAutoOffsetReset;
223.     }
225.     public void setKafkaAutoOffsetReset(String kafkaAutoOffsetReset) {
226.         this.kafkaAutoOffsetReset = kafkaAutoOffsetReset;
227.     }
229.     public void create() throws IOException {
230.         String active = System.getenv("active");
231.         active = StringUtils.isEmpty(active) ? "dev" : active;
232.         String resource = "application-".concat(active).concat(".properties");
233.         InputStream inputStream = this.getClass().getClassLoader().getResourceAsStream(resource);
234.         Properties properties = new Properties();
235.         properties.load(inputStream);
236.         properties.list(System.out);
237.         this.jdbcDriver = properties.getProperty("jdbc.driver");
238.         this.jdbcUrl = properties.getProperty("jdbc.url");
239.         this.jdbcUsername = properties.getProperty("jdbc.username");
240.         this.jdbcPassword = properties.getProperty("jdbc.password");
241.         this.jdbcDruidInitialSize = Integer.parseInt(properties.getProperty("jdbc.druidInitialSize"));
242.         this.jdbcDruidMaxActive = Integer.parseInt(properties.getProperty("jdbc.druidMaxActive"));
243.         this.jdbcDruidMinIdle = Integer.parseInt(properties.getProperty("jdbc.druidMinIdle"));
244.         this.jdbcDruidMaxWait = Integer.parseInt(properties.getProperty("jdbc.druidMaxWait"));
245.         this.jdbcDruidTimeBetweenEvictionRunsMillis = Long.parseLong(properties.getProperty("jdbc.druidTimeBetweenEvictionRunsMillis"));
246.         this.jdbcDruidMaxEvictableIdleTimeMillis = Long.parseLong(properties.getProperty("jdbc.druidMaxEvictableIdleTimeMillis"));
247.         this.jdbcDruidMinEvictableIdleTimeMillis = Long.parseLong(properties.getProperty("jdbc.druidMinEvictableIdleTimeMillis"));
248.         this.jdbcDruidTestWhileIdle = Boolean.valueOf(properties.getProperty("jdbc.druidTestWhileIdle"));
249.         this.jdbcDruidTestOnBorrow = Boolean.valueOf(properties.getProperty("jdbc.druidTestOnBorrow"));
250.         this.jdbcDruidTestOnReturn = Boolean.valueOf(properties.getProperty("jdbc.druidTestOnReturn"));
251.         this.jdbcDruidPoolPreparedStatements = Boolean.valueOf(properties.getProperty("jdbc.druidPoolPreparedStatements"));
252.         this.jdbcDruidMaxPoolPreparedStatementPerConnectionSize = Integer.parseInt(properties.getProperty("jdbc.druidMaxPoolPreparedStatementPerConnectionSize"));
253.         this.jdbcDruidFilters = properties.getProperty("jdbc.druidFilters");
254.         this.jdbcDruidValidationQuery = properties.getProperty("jdbc.druidValidationQuery");
256.         this.kafkaServer = properties.getProperty("kafka.servers");
257.         this.kafkaConsumerGroupId = properties.getProperty("kafka.consumer.groupId");
258.         this.kafkaKeySerializer = properties.getProperty("kafka.key.serializer");
259.         this.kafkaValueSerializer = properties.getProperty("kafka.value.serializer");
260.         this.kafkaUserEventTopic = properties.getProperty("kafka.user.event.topic");
261.         this.kafkaAutoOffsetReset = properties.getProperty("kafka.auto.offset.reset");
262.     }
263. }

复制代码

主体函数
主要实现读取kafka消息，使用map和json把消息转换为Student对象数据流
创建10秒的滚动窗口聚合Student数据，末了调用自定义sink存入至于mysql

1. package com.slink;
3. import com.slink.entity.Student;
4. import com.slink.function.StudentProcessWindowFunction;
5. import com.slink.properties.SinkProperties;
6. import com.slink.util.Constant;
7. import com.slink.util.KafkaSourceUtil;
8. import com.slink.util.SinkFunctionUtil;
9. import com.slink.util.StreamExecutionEnvironmentUtil;
10. import org.apache.flink.streaming.api.datastream.DataStream;
11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
12. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
13. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
14. import org.apache.flink.streaming.api.windowing.time.Time;
16. public class StudentRunner {
18.     public static void main(String[] args) throws Exception {
19.         // 创建流处理执行环境
20.         final StreamExecutionEnvironment env = StreamExecutionEnvironmentUtil.buildStreamExecutionEnvironment();
21.         // 读取环境变量 jdbc、kafka
22.         SinkProperties properties = new SinkProperties();
23.         properties.create();
24.         // 构建kafka源
25.         DataStream<Student> kafkaOut = KafkaSourceUtil.buildDataStream(properties, env, Constant.TWO);
26.         // 构建SinkFunction
27.         SinkFunction<Student> sinkFunction = SinkFunctionUtil.buildSinkFunction(properties);
28.         /**
29.          * 将清洗后的数据通过窗口去聚合(每10秒滚动窗口聚合一次)写入Mysql
30.          * 在数据清洗完成后，将数据写入Mysql数据库。这里我们设置了写入Mysql的并行度为2(setParallelism(2))。
31.          * 这意味着将有2个并发任务负责将数据写入到Mysql。由于Mysql的写入通常涉及磁盘I/O操作，设置较低的并行度可以避免I/O争用
32.          */
33.         // 全局10秒滚动窗口
34.         kafkaOut.rebalance().windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
35.                 .process(new StudentProcessWindowFunction())
36.                 .disableChaining()
37.                 .addSink(sinkFunction)
38.                 .name("KafkaFoMysql");
39. //                .setParallelism(2);
40.         kafkaOut.print(); //调度输出
41.         env.execute("flink kafka to Mysql");
42.     }
43. }

复制代码

StreamExecutionEnvironmentUtil：

1. package com.slink.util;
3. import org.apache.flink.streaming.api.CheckpointingMode;
4. import org.apache.flink.streaming.api.environment.CheckpointConfig;
5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
7. public class StreamExecutionEnvironmentUtil {
9.     private static final String statebackend_address = "file:/Users/xumingzhong/Desktop/xmz";
11.     public static StreamExecutionEnvironment buildStreamExecutionEnvironment(){
12.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
13.         /**
14.          * 设置全局并行度（默认并行度）所有算子，默认的并行度就都为6。一般不会在程序中设置全局并行度。
15.          * 因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。
16.          * 注意：由于keyBy不是算子，所以无法对keyBy设置并行度
17.          */
18. //        env.setParallelism(1);
19.         //每隔10s进行启动一个检查点【设置checkpoint的周期】
20.         env.enableCheckpointing(10000);
21.         //设置EXACTLY_ONCE语义,默认就是这个
22.         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
23.         //确保检查点之间有1s的时间间隔【checkpoint最小间隔】
24.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
25.         //检查点必须在60s之内完成，或者被丢弃【checkpoint超时时间】
26.         env.getCheckpointConfig().setCheckpointTimeout(60000);
27.         //同一时间只允许进行一次检查点
28.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
29.         //表示一旦Flink程序被cancel后，会保留checkpoint数据，以便根据实际需要恢复到指定的checkpoint
30.         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
31.         //设置statebackend,将检查点保存在hdfs上面，默认保存在内存中。这里先保存到本地
32. //        env.setStateBackend(new FsStateBackend(statebackend_address));
33.         return env;
34.     }
35. }

复制代码

SinkFunctionUtil：

1. package com.slink.util;
3. import com.slink.entity.Student;
4. import com.slink.properties.SinkProperties;
5. import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
6. import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
7. import org.apache.flink.connector.jdbc.JdbcSink;
8. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
10. import java.sql.PreparedStatement;
12. public class SinkFunctionUtil {
14.     /**
15.      * 插入Student表SQL语句
16.      */
17.     private static final String STUDENT_INSERT = "insert into student(name1, name2, name3, name4, name5, name6, name7, name8, name9, name10, name11, name12, name13, name14, name15, name16, name17, name18, name19, name20, name21, name22, name23, name24, name25, name26, name27, name28, name29, name30, name31, name32, name33, name34, name35, name36, name37, name38, name39, name40, name41, name42, name43, name44, name45, name46, name47, name48, name49, name50) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
19.     /**
20.      * 构建SinkFunction
21.      */
22.     public static SinkFunction<Student> buildSinkFunction(SinkProperties properties){
23.         return JdbcSink.sink(STUDENT_INSERT,   // SQL 插入语句
24.                 (PreparedStatement ps, Student student) -> {
25.                     ps.setString(1, student.getName1());
26.                     ps.setString(2, student.getName2());
27.                     ps.setString(3, student.getName3());
28.                     ps.setString(4, student.getName4());
29.                     ps.setString(5, student.getName5());
30.                     ps.setString(6, student.getName6());
31.                     ps.setString(7, student.getName7());
32.                     ps.setString(8, student.getName8());
33.                     ps.setString(9, student.getName9());
34.                     ps.setString(10, student.getName10());
35.                     ps.setString(11, student.getName11());
36.                     ps.setString(12, student.getName12());
37.                     ps.setString(13, student.getName13());
38.                     ps.setString(14, student.getName14());
39.                     ps.setString(15, student.getName15());
40.                     ps.setString(16, student.getName16());
41.                     ps.setString(17, student.getName17());
42.                     ps.setString(18, student.getName18());
43.                     ps.setString(19, student.getName19());
44.                     ps.setString(20, student.getName20());
45.                     ps.setString(21, student.getName21());
46.                     ps.setString(22, student.getName22());
47.                     ps.setString(23, student.getName23());
48.                     ps.setString(24, student.getName24());
49.                     ps.setString(25, student.getName25());
50.                     ps.setString(26, student.getName26());
51.                     ps.setString(27, student.getName27());
52.                     ps.setString(28, student.getName28());
53.                     ps.setString(29, student.getName29());
54.                     ps.setString(30, student.getName30());
55.                     ps.setString(31, student.getName31());
56.                     ps.setString(32, student.getName32());
57.                     ps.setString(33, student.getName33());
58.                     ps.setString(34, student.getName34());
59.                     ps.setString(35, student.getName35());
60.                     ps.setString(36, student.getName36());
61.                     ps.setString(37, student.getName37());
62.                     ps.setString(38, student.getName38());
63.                     ps.setString(39, student.getName39());
64.                     ps.setString(40, student.getName40());
65.                     ps.setString(41, student.getName41());
66.                     ps.setString(42, student.getName42());
67.                     ps.setString(43, student.getName43());
68.                     ps.setString(44, student.getName44());
69.                     ps.setString(45, student.getName45());
70.                     ps.setString(46, student.getName46());
71.                     ps.setString(47, student.getName47());
72.                     ps.setString(48, student.getName48());
73.                     ps.setString(49, student.getName49());
74.                     ps.setString(50, student.getName50());
75.                 },
76.                 getJdbcExecutionOptions(properties),
77.                 getJdbcConnectionOptions(properties)
78.         );
79.     }
81.     /**
82.      * 设置jdbc批处理
83.      */
84.     private static JdbcExecutionOptions getJdbcExecutionOptions(SinkProperties properties) {
85.         return JdbcExecutionOptions.builder()
86.                 .withBatchSize(5000)            // 设置批量插入大小
87.                 .withBatchIntervalMs(200)       // 设置批量插入的时间间隔-毫秒
88.                 .withMaxRetries(3)            // 设置最大重试次数
89.                 .build();
90.     }
92.     /**
93.      * 设置jdbc连接
94.      */
95.     public static JdbcConnectionOptions getJdbcConnectionOptions(SinkProperties properties){
96.         return new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
97.                 .withUrl(properties.getJdbcUrl())
98.                 .withDriverName(properties.getJdbcDriver())
99.                 .withUsername(properties.getJdbcUsername())
100.                 .withPassword(properties.getJdbcPassword())
101.                 //.withDataSource(dataSource)   1.17.X版本支持数据池连接配置，需JDK11
102.                 .build();
103.     }
104. }

复制代码

KafkaSourceUtil：

1. package com.slink.util;
3. import com.slink.entity.SinkLog;
4. import com.slink.entity.Student;
5. import com.slink.properties.SinkProperties;
6. import org.apache.flink.api.common.functions.MapFunction;
7. import org.apache.flink.api.common.serialization.SimpleStringSchema;
8. import org.apache.flink.streaming.api.datastream.DataStream;
9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
11. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
13. import java.util.Properties;
15. public class KafkaSourceUtil {
17.     /**
18.      * 设置Kafka源
19.      * Kafka 消费并行度：通过 setParallelism(x) 为从 Kafka 读取数据的操作设置了并行度为x。
20.      * 也就是说，Flink 将会启动 x 个并行任务来从Kafka 的 kafkaTopic 主题中消费数据。
21.      * 这个并行度可以根据 Kafka 分区的数量调整。如果 Kafka 有 x 个分区，那么设置并行度为 x 是合理的，
22.      * 这样可以保证每个分区都有一个并发实例进行处理
23.      */
24.     public static DataStream<Student> buildDataStream(SinkProperties properties, StreamExecutionEnvironment env, int parallelism) {
25.         // 构建kafka环境变量对象
26.         Properties props = new Properties();
27.         props.put("bootstrap.servers", properties.getKafkaServer());
28.         props.put("group.id", properties.getKafkaConsumerGroupId());
29.         props.put("auto.offset.reset", properties.getKafkaAutoOffsetReset());
30.         return env.addSource(new FlinkKafkaConsumer<>(
31.                         properties.getKafkaUserEventTopic(),
32.                         new SimpleStringSchema(),
33.                         props)).setParallelism(parallelism)
34.                 .map(new MapFunction<String, Student>() {
35.                     @Override
36.                     public Student map(String val) {
37.                         // 解析字符串转换Student对象
38.                         return SinkLog.build(val);
39.                     }
40.                 }).filter(v -> v != null);
41.     }
42. }

复制代码

工具类：EmptyNullUtil

1. package com.slink.util;
3. import org.apache.commons.lang3.StringUtils;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
7. import java.lang.reflect.Field;
8. import java.lang.reflect.Method;
10. public class EmptyNullUtil {
12.     private static final Logger log = LoggerFactory.getLogger(EmptyNullUtil.class);
14.     /**
15.      * 将属性类型为String，值为null的设置为 ""
16.      */
17.     public static <T> void stringNullToEmpty(T t) {
18.         if (null == t) {
19.             return;
20.         }
21.         Field[] declaredFields = t.getClass().getDeclaredFields();
22.         for (Field field : declaredFields) {
23.             field.setAccessible(true);
24.             if (field.getType().equals(String.class)) {
25.                 // 将属性的首字母大写
26.                 String methodName = field.getName().replaceFirst(field.getName().substring(0, 1), field.getName().substring(0, 1).toUpperCase());
27.                 try {
28.                     Method methodGet = t.getClass().getMethod("get" + methodName);
29.                     // 调用getter方法获取属性值
30.                     String str = (String) methodGet.invoke(t);
31.                     if (StringUtils.isBlank(str)) {
32.                         // 如果为null的String类型的属性则重新复制为空字符串
33.                         field.set(t, field.getType().getConstructor(field.getType()).newInstance(StringUtils.EMPTY));
34.                     }
35.                 } catch (Exception e) {
36.                     log.warn("[EmptyNullUtil.stringBlankToNull] e:{}", e);
37.                 }
38.             }
39.         }
40.     }
41. }

复制代码

窗口函数自定义处理类：StudentProcessWindowFunction

1. package com.slink.function;
3. import com.slink.StudentRunner;
4. import com.slink.entity.Student;
5. import org.apache.commons.compress.utils.Lists;
6. import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
7. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
8. import org.apache.flink.util.Collector;
9. import org.slf4j.Logger;
10. import org.slf4j.LoggerFactory;
12. import java.util.List;
13. import java.util.stream.Collectors;
14. import java.util.stream.StreamSupport;
16. public class StudentProcessWindowFunction extends ProcessAllWindowFunction<Student, Student, TimeWindow> {
18.     private static final Logger log = LoggerFactory.getLogger(StudentRunner.class);
20.     @Override
21.     public void process(ProcessAllWindowFunction<Student, Student, TimeWindow>.Context context, Iterable<Student> iterable, Collector<Student> out) {
22.         log.info("窗口聚合数据条数:{}", StreamSupport.stream(iterable.spliterator(), Boolean.FALSE).collect(Collectors.toList()).size());
23.         // 直接将窗口内的每个 Student 输出
24.         for (Student order : iterable) {
25.             out.collect(order);
26.         }
27.     }
28. }

复制代码

四、运行测试

模拟数据每秒推送一条数据至Kafka:

1. package com.slink;
3. import org.apache.kafka.clients.producer.KafkaProducer;
4. import org.apache.kafka.clients.producer.ProducerConfig;
5. import org.apache.kafka.clients.producer.ProducerRecord;
6. import org.apache.kafka.common.serialization.StringSerializer;
8. import java.util.Properties;
10. public class Test2 {
12.     public static void main(String[] args) {
13.         String value = "{ "timestamp": "2024-11-05T17:38:56+08:00", "remoteAddr": "61.132.71.10","costime": "0.000","realtime": "0.000","status": 405,"xForwarded": "","referer": "","request": "POST /sink HTTP/1.1","upstrAddr": "127.0.0.1:80","bytes":157,"requestBody":{"name39":"拓扑丝路39号","name38":"拓扑丝路38号","name37":"拓扑丝路37号","name36":"拓扑丝路36号","name35":"拓扑丝路35号","name34":"拓扑丝路34号","name33":"拓扑丝路33号","name32":"拓扑丝路32号","name31":"拓扑丝路31号","name30":"拓扑丝路30号","name29":"拓扑丝路29号","name28":"拓扑丝路28号","name27":"拓扑丝路27号","name6":"拓扑丝路6号","name26":"拓扑丝路26号","name5":"拓扑丝路5号","name25":"拓扑丝路25号","name4":"拓扑丝路4号","name24":"拓扑丝路24号","name3":"拓扑丝路3号","name23":"拓扑丝路23号","name22":"拓扑丝路22号","name9":"拓扑丝路9号","name21":"拓扑丝路21号","name8":"拓扑丝路8号","name20":"拓扑丝路20号","name7":"拓扑丝路7号","name2":"拓扑丝路2号","name1":"拓扑丝路1号","name19":"拓扑丝路19号","name18":"拓扑丝路18号","name17":"拓扑丝路17号","name16":"拓扑丝路16号","name15":"拓扑丝路15号","name14":"拓扑丝路14号","name13":"拓扑丝路13号","name12":"拓扑丝路12号","name11":"拓扑丝路11号","name10":"拓扑丝路10号","name50":"拓扑丝路50号","name49":"拓扑丝路49号","name48":"拓扑丝路48号","name47":"拓扑丝路47号","name46":"拓扑丝路46号","name45":"拓扑丝路45号","name44":"拓扑丝路44号","name43":"拓扑丝路43号","name42":"拓扑丝路42号","name41":"拓扑丝路41号","name40":"拓扑丝路40号"},"agent": "PostmanRuntime/7.42.0" }";
14.         //创建生产者
15.         Properties properties = new Properties();
16.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
17.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
18.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
19.         //优化参数
20.         properties.put(ProducerConfig.BATCH_SIZE_CONFIG,1024*1024);//生产者尝试缓存记录，为每一个分区缓存一个mb的数据
21.         properties.put(ProducerConfig.LINGER_MS_CONFIG,500);//最多等待0.5秒.
22.         KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
23.         for(int i=0; i<10000; i++){
24.             ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "key001", value);
25.             kafkaProducer.send(record);
26.             try {
27.                 Thread.sleep(1000);
28.             } catch (InterruptedException e) {
29.                 e.printStackTrace();
30.             }
32.         }
33.         kafkaProducer.flush();
34.         kafkaProducer.close();
35.     }
36. }

复制代码

运行主体函数：StudentRunner

查询Mysql日记

​​​​​​​
实例运行乐成
五、总结

本文实例了实现了从Kafka及时读取数据，根据定制化处理数据，通过flink窗口模式批量写入数据库mysql，可根据自身需求写入其他存储(ES、Redis等)。该实例适合在对数据库及时性要求不高，大概是准及时数据分析时的场景。如若数据量大的环境下，聚合十秒钟数据达万条，那么这样批量写会比单条性能进步很多倍。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。