flink连接kafka

1. 使用简单的 Flink 测试作业

编写一个简单的 Flink 测试作业，尝试从 Kafka 中读取数据或将数据写入 Kafka。这可以帮助验证 Flink 是否能够与 Kafka 进行基本的通信。

示例代码：读取 Kafka 数据

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

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

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.23.60:9092")
                .setTopics("test_topic")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStream<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");

        stream.print(); // Print the data to the console for debugging

        env.execute("Kafka Read Test");
    }
}

示例代码：写入 Kafka 数据

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

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

        DataStream<String> stream = env.fromElements("test message 1", "test message 2");

        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.23.60:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("test_topic")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                .build();

        stream.sinkTo(kafkaSink);

        env.execute("Kafka Write Test");
    }
}

2. 使用 Kafka 工具检查

使用 Kafka 的工具验证数据是否到达了 Kafka 主题。

生产者测试：使用 Kafka 的 kafka-console-producer 工具向主题写入测试数据，确保数据能够成功写入 Kafka。
1
kafka-console-producer.sh --broker-list 192.168.23.60:9092 --topic test_topic
消费者测试：使用 Kafka 的 kafka-console-consumer 工具从主题读取数据，确保数据能够成功读取。
1
kafka-console-consumer.sh --bootstrap-server 192.168.23.60:9092 --topic test_topic --from-beginning