使用两个过程访问Kafka主题

Question

我有一个可以正常工作的kafka生产者类。制作人填补了Kafka主题。它的代码如下：

public class kafka_test {
private final static String TOPIC = "flinkTopic";
private final static String BOOTSTRAP_SERVERS = "10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092";
public FlinkKafkaConsumer<String> createStringConsumerForTopic(
        String topic, String kafkaAddress, String kafkaGroup) {
    //        ************************** KAFKA Properties ******
    Properties props = new Properties();
    props.setProperty("bootstrap.servers", kafkaAddress);
    props.setProperty("group.id", kafkaGroup);
    FlinkKafkaConsumer<String> myconsumer = new FlinkKafkaConsumer<>(
            topic, new SimpleStringSchema(), props);
    myconsumer.setStartFromLatest();
    return myconsumer;
}
  private static Producer<Long, String> createProducer() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
    props.put(ProducerConfig.CLIENT_ID_CONFIG, "MyKafkaProducer");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    return new KafkaProducer<>(props);
}
public void runProducer(String msg) throws Exception {
    final Producer<Long, String> producer = createProducer();
    try {
            final ProducerRecord<Long, String> record = new ProducerRecord<>(TOPIC, msg );
            RecordMetadata metadata = producer.send(record).get();
            System.out.printf("sent record(key=%s value='%s')" + " metadata(partition=%d, offset=%d)n",
                    record.key(), record.value(), metadata.partition(), metadata.offset());
    } finally {
        producer.flush();
        producer.close();
    }
 }
}
  public class producerTest {
  public static void main(String[] args) throws Exception{
    kafka_test objKafka=new kafka_test();
    String pathFile="/home/cfms11/IdeaProjects/pooyaflink2/KafkaTest/quickstart/lastDay4.csv";
    String delimiter="n";
   objKafka.createStringProducer("flinkTopic",
   "10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092");
    Scanner scanner = new Scanner(new File(pathFile));
    scanner.useDelimiter(delimiter);
    int i=0;
    while(scanner.hasNext()){
        if (i==0)
            TimeUnit.MINUTES.sleep(1);
         objKafka.runProducer(scanner.next());
       i++;
    }
    scanner.close();
    }
   }

因为，我想为我的Flink程序提供数据，因此，我使用Kafka。实际上，我有此部分代码可以从Kafka主题中消耗数据：

    Properties props = new Properties();
    props.setProperty("bootstrap.servers", 
    "10.32.0.2:9092,10.32.0.3:9092,10.32.0.4:9092");
    props.setProperty("group.id", kafkaGroup);
    FlinkKafkaConsumer<String> myconsumer = new FlinkKafkaConsumer<>(
            "flinkTopic", new SimpleStringSchema(), props);
    DataStream<String> text =   env.addSource(myconsumer).setStartFromEarliest());

我想在程序运行的同时运行生产者代码。我的目标是，生产者将一个记录发送到该主题，消费者可以同时通过主题记录该记录。

您能告诉我如何以及如何管理它。

Answer 1

我认为您需要创建两个类文件，一个是生产者，另一个是消费者。首先创建主题，然后运行消费者，或直接运行生产者。