Flume有两种方式给消息输入到Kafka中,一种是Kafka channel,另外一种是Kafka sink .

Kafka Sink

步骤:

1）配置flume(flume-kafka.conf)

  1. # define
  2. a1.sources = r1
  3. a1.sinks = k1
  4. a1.channels = c1
  6. # source
  7. a1.sources.r1.type = exec
  8. a1.sources.r1.command = tail -F -c +0 /opt/module/datas/flume.log
  9. a1.sources.r1.shell = /bin/bash -c
  11. # sink
  12. a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  13. a1.sinks.k1.kafka.bootstrap.servers = zjj101:9092,zjj102:9092,zjj103:9092
  14. a1.sinks.k1.kafka.topic = first
  15. # 一次批量发送20个
  16. a1.sinks.k1.kafka.flumeBatchSize = 20
  17. a1.sinks.k1.kafka.producer.acks = 1
  18. a1.sinks.k1.kafka.producer.linger.ms = 1
  20. # channel
  21. a1.channels.c1.type = memory
  22. a1.channels.c1.capacity = 1000
  23. a1.channels.c1.transactionCapacity = 100
  25. # bind
  26. a1.sources.r1.channels = c1
  27. a1.sinks.k1.channel = c1

2） 启动kafkaIDEA消费者

  1. package com.consumer;
  3. import org.apache.kafka.clients.consumer.ConsumerRecord;
  4. import org.apache.kafka.clients.consumer.ConsumerRecords;
  5. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import java.util.Arrays;
  8. import java.util.Properties;
  10. public class CustomConsumer {
  11.     public static void main(String[] args) {
  14.         Properties props = new Properties();
  16.         props.put("bootstrap.servers", "zjj101:9092");
  18.         props.put("group.id", "test");
  19.         //enable.auto.commit：是否开启自动提交offset功能
  20.         props.put("enable.auto.commit", "true");
  21.         //auto.commit.interval.ms：自动提交offset的时间间隔
  22.         props.put("auto.commit.interval.ms", "1000");
  24.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  25.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  27.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  28.         //订阅 名字为 first的  topic ,可以订阅好几个topic
  29.         consumer.subscribe(Arrays.asList("first"));
  31.         while (true) {
  32.             //拉取时间,如果拉取时间超过这个时间还没拉取出来数据就算拉失败了
  33.             ConsumerRecords<String, String> records = consumer.poll(100);
  35.             for (ConsumerRecord<String, String> record : records) {
  36.                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
  37.                 //输出: offset = 400, key = 0, value = 0
  38.             }
  39.         }
  40.     }
  41. }

3） 进入flume根目录下，启动flume


$ bin/flume-ng agent -c conf/ -n a1 -f jobs/flume-kafka.conf

4） 向 /opt/module/datas/flume.log里追加数据，查看kafka消费者消费情况

$ echo hello >> /opt/module/datas/flume.log