1. 启动zookeeper、kafka并创建kafka主题

./bin/zkServer.sh start

./bin/kafka-server-start.sh -daemon ./config/server.properties./bin/kafka-topic.sh --create --topic hunter --partitions 3 --replication-factor 1 --zookeeper localhost:9092

2、创建flume-kafka.conf配置文件

用于采集socket数据后存入kafka

在flume文件夹中的conf下新建flume-kafka.conf配置文件

vim flume-kafka.conf

a1.sources = s1
a1.channels = c1
a1.sinks = k1a1.sources.s1.type = netcat
a1.sources.s1.bind = localhost 
a1.sources.s1.port = 10050 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = hunter
a1.sinks.k1.kafka.bootstrap.servers = master:9092a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1

设置监听本地端口10050 netcat发送的socket数据，讲采集到的数据存入kafka的hunter主题中

3、启动flume

./bin/flue-ng agent -n a1 -c conf -f ./conf/flume-kafka.conf -Dflume.root.logger=INFO,console

./bin/flume-ng：启动Flume-ng二进制文件。

agent：指定要启动的Flume组件类型为代理。

-c conf：设置Flume配置文件所在的目录为当前目录下的conf目录。

-f ./conf/flume_kafka.conf：指定Flume代理使用的配置文件路径和名称。

-n a1：给Flume代理指定一个名称为a1。

-Dflume.root.logger=INFO,console：设置Flume代理的日志级别为INFO，并将日志输出到控制台。

4、创建kafka消费者

./bin/kafka-console-consumer.sh --from-beginning --topic hunter --bootstrap-server localhost:9092

/bin/kafka-console-consumer.sh：启动 Kafka 控制台消费者。

--from-beginning：从该主题的开始位置读取消息。

--topic hunter：指定要消费的主题名称为 "hunter"。

--bootstrap-server 集群各个主机:端口：指定连接到 Kafka 集群的所有 broker 的主机名和端口号

5、netcat向本地10050端口发送socket数据

先开启flume，这里就会直接进入输入模式

nc localhost 10050

读取成功示例