双十二大数据消息处理推荐

// Kafka Producer示例
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<>("transaction-topic", "key", "value"));
producer.close();

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

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

stream.map(new MapFunction<String, Transaction>() {
    @Override
    public Transaction map(String value) throws Exception {
        return JSON.parseObject(value, Transaction.class);
    }
}).keyBy(Transaction::getUserId)
 .timeWindow(Time.seconds(10))
 .aggregate(new AggregateFunction<Transaction, Double, Double>() {
     @Override
     public Double createAccumulator() {
         return 0.0;
     }

     @Override
     public Double add(Transaction value, Double accumulator) {
         return accumulator + value.getAmount();
     }

     @Override
     public Double getResult(Double accumulator) {
         return accumulator;
     }

     @Override
     public Double merge(Double a, Double b) {
         return a + b;
     }
 }).print();

env.execute("Transaction Aggregation");