双11大数据消息处理购买

// Kafka Producer Example
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<>("order-topic", orderId, orderDetails));
producer.close();

// Flink Consumer Example
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties consumerProps = new Properties();
consumerProps.setProperty("bootstrap.servers", "localhost:9092");
consumerProps.setProperty("group.id", "order-group");

DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("order-topic", new SimpleStringSchema(), consumerProps));

stream.map(new MapFunction<String, Order>() {
    @Override
    public Order map(String value) throws Exception {
        return parseOrder(value); // 自定义解析逻辑
    }
}).print();

env.execute("Order Processing Job");