大数据消息处理限时活动

// Kafka生产者示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key", "value"));
producer.close();

// Flink消费者示例
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties consumerProps = new Properties();
consumerProps.setProperty("bootstrap.servers", "localhost:9092");
consumerProps.setProperty("group.id", "flink-consumer-group");

DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), consumerProps));
stream.map(new MapFunction<String, String>() {
    @Override
    public String map(String value) throws Exception {
        return "Processed: " + value;
    }
}).print();

env.execute("Flink Kafka Example");