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>
6
7    <groupId>org.example</groupId>
8    <artifactId>flink_sdk</artifactId>
9    <version>1.0-SNAPSHOT</version>
10
11    <properties>
12        <maven.compiler.source>1.8</maven.compiler.source>
13        <maven.compiler.target>1.8</maven.compiler.target>
14        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15    </properties>
16
17    <dependencies>
18        <dependency>
19            <groupId>org.apache.kafka</groupId>
20            <artifactId>kafka-clients</artifactId>
21            <version>0.10.2.2</version>
22        </dependency>
23        <dependency>
24            <groupId>org.slf4j</groupId>
25            <artifactId>slf4j-simple</artifactId>
26            <version>1.7.25</version>
27            <scope>compile</scope>
28        </dependency>
29        <dependency>
30            <groupId>org.apache.flink</groupId>
31            <artifactId>flink-java</artifactId>
32            <version>1.6.1</version>
33        </dependency>
34        <dependency>
35            <groupId>org.apache.flink</groupId>
36            <artifactId>flink-streaming-java_2.11</artifactId>
37            <version>1.6.1</version>
38        </dependency>
39        <dependency>
40            <groupId>org.apache.flink</groupId>
41            <artifactId>flink-connector-kafka_2.11</artifactId>
42            <version>1.7.0</version>
43        </dependency>
44    </dependencies>
45
46    <build>
47        <plugins>
48            <plugin>
49                <groupId>org.apache.maven.plugins</groupId>
50                <artifactId>maven-shade-plugin</artifactId>
51                <version>3.2.4</version>
52                <executions>
53                    <execution>
54                        <phase>package</phase>
55                        <goals>
56                            <goal>shade</goal>
57                        </goals>
58                        <configuration>
59                            <transformers>
60                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
61                                    <mainClass>org.example.KafkaConsumerDemo</mainClass>
62                                </transformer>
63                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
64                                    <resource>reference.conf</resource>
65                                </transformer>
66                            </transformers>
67                        </configuration>
68                    </execution>
69                </executions>
70            </plugin>
71        </plugins>
72    </build>
73
74</project>

1package org.example;
2
3import org.apache.kafka.clients.producer.KafkaProducer;
4import org.apache.kafka.clients.producer.ProducerConfig;
5import org.apache.kafka.clients.producer.ProducerRecord;
6import org.apache.kafka.clients.producer.RecordMetadata;
7import org.apache.kafka.common.serialization.StringSerializer;
8
9import java.util.Properties;
10
11public class KafkaProducerDemo {
12    public static void main(String args[]) throws Exception {
13
14        //接入点设置
15        String access_point = "x.x.x.x:9092,x.x.x.x:9092,x.x.x.x:9092";
16        //填写需要发送消息的主题名称
17        String topic = "topic_name";
18        //填写需要发送消息
19        String value = "test flink message";
20        // 创建配置类
21        Properties props = new Properties();;
22        // 指定kafka服务端所在位置
23        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, access_point);
24        // Kafka消息的序列化方式。
25        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
26        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
27        // 请求的最长等待时间。
28        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
29        // 设置客户端内部重试次数。
30        props.put(ProducerConfig.RETRIES_CONFIG, 5);
31        // 设置客户端内部重试间隔。
32        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
33        // 构造 Producer 对象
34        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
35
36        try {
37            // 测试发送 100 条消息
38            for (int i = 0; i < 100; i++) {
39                ProducerRecord<String, String> kafkaMessage =  new ProducerRecord<>(topic, value + ": " + i);
40                kafkaProducer.send(kafkaMessage, (RecordMetadata recordMetadata, Exception e) -> {
41                    if (e == null) {
42                        System.out.println("send success:" + recordMetadata.toString());
43                    } else {
44                        e.printStackTrace();
45                        System.err.println("send fail");
46                    }
47                });
48            }
49        } catch (Exception e) {
50            System.out.println("Something error has happened");
51            e.printStackTrace();
52        } finally {
53            kafkaProducer.close();
54        }
55    }
56}

1package org.example;
2
3import org.apache.flink.api.common.serialization.SimpleStringSchema;
4import org.apache.flink.streaming.api.datastream.DataStream;
5import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
6import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
7
8import java.util.Properties;
9
10
11public class KafkaConsumerDemo {
12    public static void main(String args[]) throws Exception {
13
14        //接入点设置
15        String access_point = "x.x.x.x:9092,x.x.x.x:9092,x.x.x.x:9092";
16        String group_id = "flink_group";
17        
18        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
19        // 创建配置类
20        Properties properties = new Properties();
21        // 设置接入点
22        properties.setProperty("bootstrap.servers", access_point);
23        // 设置消费组id
24        properties.setProperty("group.id", group_id);
25        // FlinkKafkaConsumer的第一个参数填创建的topic名称
26        DataStream<String> stream = env
27                .addSource(new FlinkKafkaConsumer<>("topic_name", new SimpleStringSchema(), properties));
28        // 打印输出
29        stream.print();
30        env.execute();
31    }
32}

1java -jar flink_sdk-1.0-SNAPSHOT.jar