flink-kafka 简单例子(java)

Posted 2021-12-03 smile-yan

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了flink-kafka 简单例子(java)相关的知识，希望对你有一定的参考价值。

问题描述

很久以前实习的时候接触flink，并且在此方面做了很多工作，但是过了几年时间，居然再次搭建 kafka-flink 会遇到那么多困难，无意间翻到那个时候写的文档，这里再次总结并公开一下，希望也能帮到需要的人。

效果描述

首先必须强调一点：以下内容并不是以 flink 的job的方式在flink中运行，但是，可以打包成 jar ，然后提交到 flink 中运行。

因为本地测试的需要，一般这种入门级别，用来验证 kafka 的项目，本地运行即可。

当启动本地 IDE，即进入监听 kafka 的状态，然后进入云服务器，启动kafka 生产者脚本，手动输入消息到 kafka 中。IDE 中的效果就是消费这些消息，全部都打印出来。如图所示：

源码

源码地址：https://gitee.com/smile-yan/flink-kafka-demo.git

环境介绍：

kafka 版本：kafka_2.13-3.0.0，但是实际上跟 kafka 版本联系不大。相差不太大的版本应该都可以用。
Oracle JDK 1.8
Flink ：flink-1.12.5 ，为了和 pom 保持此一致，最好使用 1.12 版本比较稳定，其他版本可以考虑试一下。

pom 文件，注意要保持 kafka, flink 版本相一致，不要轻易尝试随便改其中某一个版本。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.smileyan.demo</groupId>
    <artifactId>flink-kafka-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>
    <properties>
        <influxdb.java.version>2.15</influxdb.java.version>
        <okhttp.version>3.13.1</okhttp.version>
        <flink.version>1.12.1</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <maven-shade-plugin.version>3.2.4</maven-shade-plugin.version>
    </properties>
    <profiles>
        <profile>
            <id>local</id>
            <activation>
                <activeByDefault>true</activeByDefault>
            </activation>
            <properties>
                <flink.scope>compile</flink.scope>
            </properties>
        </profile>
        <profile>
            <id>prod</id>
            <activation>
                <activeByDefault>false</activeByDefault>
            </activation>
            <properties>
                <flink.scope>provided</flink.scope>
            </properties>
        </profile>
    </profiles>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>${flink.scope}</scope>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>*</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>${flink.scope}</scope>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>*</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>${flink.scope}</scope>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>*</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>*</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>*</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.69</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>${maven-shade-plugin.version}</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <createDependencyReducedPom>false</createDependencyReducedPom>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

接着一个java文件，源码如下：

package cn.smileyan.demo;

import java.util.Properties;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * 作者 smileyan
 * 参考博客：https://smileyan.blog.csdn.net/article/details/121044085
 */
public class HelloFlink {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        props.put("bootstrap.servers", "服务器地址:9092");
        props.put("zookeeper.connect", "服务器地址:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");

        DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer<String>(
                "metric",
                new SimpleStringSchema(),
                props)).setParallelism(1);

        dataStreamSource.print();
        env.execute("Flink add data source");
    }

}

运行效果如上面第一个图所示，再次说明：图中的效果是本地IDE启动后，监听 kafka 的效果，并没有真正打包成 jar 项目并提交到 flink 而运行。但是这份源码是可以打包提交到 flink 运行的（已测有效）

总结

由于技术需要，可能还会在这方面多花一些时间，主要会在流计算上花更多的时间。特意做个笔记，共勉。

Smileyan
2021.10.30 18:16

以上是关于flink-kafka 简单例子(java)的主要内容，如果未能解决你的问题，请参考以下文章

一个简单的时间片轮转内核代码的分析（课程作业）

lab2:完成一个简单的时间片轮转多道程序内核代码

用java给html文件添加必要的控制html代码片

完成一个简单的时间片轮转多道程序内核代码

Socket 多线程编程

Linux内核分析：完成一个简单的时间片轮转多道程序内核代码