文章目录
前言
一、版本
二、使用步骤
1.maven引入库
2.上代码
前言
近期开始学习Flink程序开发，使用java语言，此文以生产数据至kafka为例记录下遇到的问题以及代码实现，若有错误请提出。
一、版本
Flink版本：1.15.4
kafka版本：3.0.0
二、使用步骤
1.maven引入库
<properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <maven.compiler.encoding>UTF-8</maven.compiler.encoding> <flink.version>1.15.4</flink.version> <target.java.version>1.8</target.java.version> <scala.binary.version>2.11</scala.binary.version> <maven.compiler.source>${target.java.version}</maven.compiler.source> <maven.compiler.target>${target.java.version}</maven.compiler.target> <log4j2.version>2.14.1</log4j2.version> </properties><dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-runtime-web</artifactId> <version>${flink.version}</version> </dependency>
2.上代码
以下代码将Flink环境初始化、配置、生产数据至kafka代码放在一个独立类的原因，主要是在我近期做的业务中需要有多个实例往kafka生产数据，后面两个启动类调用此类。文章来源：https://www.toymoban.com/news/detail-425390.html
import com.alibaba.fastjson.JSON;
import com.dogame.data.report.comsume.bean.KafkaEnvParameter;
import com.dogame.data.report.comsume.bean.Order;
import com.dogame.data.report.comsume.utils.EnvironmentUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import java.util.Random;
import java.util.UUID;
public class WorkerService {
public void run (String[] args, String bizType) throws Exception {
StreamExecutionEnvironment ENV = EnvironmentUtils.initEnvironmentConfig(args, bizType);
DataStream<String> orderDataStream = ENV.addSource(new ParallelSourceFunction<String>() {
private boolean running = true;
@Override
public void run(SourceContext<String> ctx) throws Exception {
String [] channels = {"www.xxx.com","www.sdfs.com","www.qqq.com","www.dfff.com"};
while(running){
Order order = new Order();
order.setOrderId(String.valueOf(System.currentTimeMillis()+new Random().nextInt()));
order.setTime(System.currentTimeMillis());
order.setFee(1000* new Random().nextDouble());
order.setChannel(channels[new Random().nextInt(channels.length)]);
ctx.collect(JSON.toJSONString(order)); //发送出去
Thread.sleep(10);
}
}
@Override
public void cancel() {
running = false;
}
});
orderDataStream.print();
KafkaSink<String> sink = KafkaSink.<String>builder()
.setKafkaProducerConfig(KafkaEnvParameter.productProperties)
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic(KafkaEnvParameter.responseTopics)
.setValueSerializationSchema(new SimpleStringSchema())
.build()
)
.setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
//.setTransactionalIdPrefix(UUID.randomUUID().toString())
.build();
orderDataStream.sinkTo(sink);
orderDataStream.print();
ENV.execute("SourceFunctionTest");
}
}
@Slf4j
public class SourceFunctionTest {
public static void main(String[] args) {
try {
String bizType = "t0";
WorkerService workerService = new WorkerService();
workerService.run(args, bizType);
} catch (Exception e) {
log.error(" fail ", e);
}
}
}
@Slf4j
public class SourceFunctionTest1 {
public static void main(String[] args) throws Exception {
try {
String bizType = "t1";
WorkerService workerService = new WorkerService();
workerService.run(args, bizType);
} catch (Exception e) {
log.error(" fail ", e);
}
}
}需要强调一点的是：经过测试，发现以下问题：若只启动一个实例，代码能稳定运行，若启动两个以上实例，运行几分钟后会有程序报错，最终只有一个实例能持续正常运行，经过2天对这个问题的排查（问chatGPT结合相关代码走读），发现居然是KafkaSinkBuilder这个方法未调用(以上workerService类已注释，打开即可)：文章来源地址https://www.toymoban.com/news/detail-425390.html
/**
* Sets the prefix for all created transactionalIds if {@link DeliveryGuarantee#EXACTLY_ONCE} is
* configured.
*
* <p>It is mandatory to always set this value with {@link DeliveryGuarantee#EXACTLY_ONCE} to
* prevent corrupted transactions if multiple jobs using the KafkaSink run against the same
* Kafka Cluster. The default prefix is {@link #transactionalIdPrefix}.
*
* <p>The size of the prefix is capped by {@link #MAXIMUM_PREFIX_BYTES} formatted with UTF-8.
*
* <p>It is important to keep the prefix stable across application restarts. If the prefix
* changes it might happen that lingering transactions are not correctly aborted and newly
* written messages are not immediately consumable until the transactions timeout.
*
* @param transactionalIdPrefix
* @return {@link KafkaSinkBuilder}
*/
public KafkaSinkBuilder<IN> setTransactionalIdPrefix(String transactionalIdPrefix) {
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
checkState(
transactionalIdPrefix.getBytes(StandardCharsets.UTF_8).length
<= MAXIMUM_PREFIX_BYTES,
"The configured prefix is too long and the resulting transactionalId might exceed Kafka's transactionalIds size.");
return this;
}到了这里，关于Flink生产数据到kafka的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
