kafka producer consumer 实时读数据

5icode.top

2463人浏览 · 2016-06-29 20:36:53

5icode.top · 2016-06-29 20:36:53 发布

[root@cdh2 kafka_2.9.2-0.8.1.1]# cd ../zookeeper-3.4.6/
[root@cdh2 zookeeper-3.4.6]# bin/zkServer.sh status
JMX enabled by default
Using config: /user/local/zookeeper-3.4.6/bin/../conf/zoo.cfg
Mode: standalone
[root@cdh2 zookeeper-3.4.6]# bin/zkCli.sh -server 127.0.0.1:2181
Connecting to 127.0.0.1:2181
2016-06-29 02:14:47,678 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
2016-06-29 02:14:47,681 [myid:] - INFO  [main:Environment@100] - Client environment:host.name=localhost
2016-06-29 02:14:47,681 [myid:] - INFO  [main:Environment@100] - Client environment:java.version=1.7.0_67
2016-06-29 02:14:47,683 [myid:] - INFO  [main:Environment@100] - Client environment:java.vendor=Oracle Corporation
2016-06-29 02:14:47,683 [myid:] - INFO  [main:Environment@100] - Client environment:java.home=/user/local/jdk/jre
2016-06-29 02:14:47,683 [myid:] - INFO  [main:Environment@100] - Client environment:java.class.path=/user/local/zookeeper-3.4.6/bin/../build/classes:/user/local/zookeeper-3.4.6/bin/../build/lib/*.jar:/user/local/zookeeper-3.4.6/bin/../lib/slf4j-log4j12-1.6.1.jar:/user/local/zookeeper-3.4.6/bin/../lib/slf4j-api-1.6.1.jar:/user/local/zookeeper-3.4.6/bin/../lib/netty-3.7.0.Final.jar:/user/local/zookeeper-3.4.6/bin/../lib/log4j-1.2.16.jar:/user/local/zookeeper-3.4.6/bin/../lib/jline-0.9.94.jar:/user/local/zookeeper-3.4.6/bin/../zookeeper-3.4.6.jar:/user/local/zookeeper-3.4.6/bin/../src/java/lib/*.jar:/user/local/zookeeper-3.4.6/bin/../conf:.:/user/local/jdk/lib:/user/local/jdk/jre/lib
2016-06-29 02:14:47,683 [myid:] - INFO  [main:Environment@100] - Client environment:java.library.path=/usr/java/packages/lib/i386:/lib:/usr/lib
2016-06-29 02:14:47,683 [myid:] - INFO  [main:Environment@100] - Client environment:java.io.tmpdir=/tmp
2016-06-29 02:14:47,684 [myid:] - INFO  [main:Environment@100] - Client environment:java.compiler=<NA>
2016-06-29 02:14:47,685 [myid:] - INFO  [main:Environment@100] - Client environment:os.name=Linux
2016-06-29 02:14:47,685 [myid:] - INFO  [main:Environment@100] - Client environment:os.arch=i386
2016-06-29 02:14:47,686 [myid:] - INFO  [main:Environment@100] - Client environment:os.version=2.6.32-358.el6.i686
2016-06-29 02:14:47,686 [myid:] - INFO  [main:Environment@100] - Client environment:user.name=root
2016-06-29 02:14:47,686 [myid:] - INFO  [main:Environment@100] - Client environment:user.home=/root
2016-06-29 02:14:47,686 [myid:] - INFO  [main:Environment@100] - Client environment:user.dir=/user/local/zookeeper-3.4.6
2016-06-29 02:14:47,687 [myid:] - INFO  [main:ZooKeeper@438] - Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@e13c84
Welcome to ZooKeeper!
2016-06-29 02:14:47,887 [myid:] - INFO  [main-SendThread(127.0.0.1:2181):ClientCnxn$SendThread@975] - Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
JLine support is enabled
2016-06-29 02:14:47,905 [myid:] - INFO  [main-SendThread(127.0.0.1:2181):ClientCnxn$SendThread@852] - Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
2016-06-29 02:14:47,931 [myid:] - WARN  [main-SendThread(127.0.0.1:2181):ClientCnxnSocket@139] - Connected to an old server; r-o mode will be unavailable
2016-06-29 02:14:47,931 [myid:] - INFO  [main-SendThread(127.0.0.1:2181):ClientCnxn$SendThread@1235] - Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x1559b5b0ab3000c, negotiated timeout = 30000

WATCHER::

WatchedEvent state:SyncConnected type:None path:null
[zk: 127.0.0.1:2181(CONNECTED) 0] ls /
[consumers, config, controller, admin, brokers, zookeeper, controller_epoch]
[zk: 127.0.0.1:2181(CONNECTED) 1] ls /brokers/topics
[test, lyc.gd4, zerg.hydra]
[zk: 127.0.0.1:2181(CONNECTED) 2] delete /brokers/topics/zerg.hydra
Node not empty: /brokers/topics/zerg.hydra
[zk: 127.0.0.1:2181(CONNECTED) 3] deleteall /brokers/topics/zerg.hydra
ZooKeeper -server host:port cmd args
	connect host:port
	get path [watch]
	ls path [watch]
	set path data [version]
	rmr path
	delquota [-n|-b] path
	quit 
	printwatches on|off
	create [-s] [-e] path data acl
	stat path [watch]
	close 
	ls2 path [watch]
	history 
	listquota path
	setAcl path acl
	getAcl path
	sync path
	redo cmdno
	addauth scheme auth
	delete path [version]
	setquota -n|-b val path
[zk: 127.0.0.1:2181(CONNECTED) 4] delete /brokers/topics/zerg.hydra   
Node not empty: /brokers/topics/zerg.hydra
[zk: 127.0.0.1:2181(CONNECTED) 5] ls /brokers/topics                  
[test, lyc.gd4, zerg.hydra]
[zk: 127.0.0.1:2181(CONNECTED) 6] ls /brokers/topics
[test, lyc.gd4, zerg.hydra]
[zk: 127.0.0.1:2181(CONNECTED) 7] delete /brokers/topics/zerg.hydra
Node not empty: /brokers/topics/zerg.hydra
[zk: 127.0.0.1:2181(CONNECTED) 8] deleteall /brokers/topics/zerg.hydra
ZooKeeper -server host:port cmd args
	connect host:port
	get path [watch]
	ls path [watch]
	set path data [version]
	rmr path
	delquota [-n|-b] path
	quit 
	printwatches on|off
	create [-s] [-e] path data acl
	stat path [watch]
	close 
	ls2 path [watch]
	history 
	listquota path
	setAcl path acl
	getAcl path
	sync path
	redo cmdno
	addauth scheme auth
	delete path [version]
	setquota -n|-b val path
[zk: 127.0.0.1:2181(CONNECTED) 9] delete /brokers/topics/zerg.hydra   
Node not empty: /brokers/topics/zerg.hydra
[zk: 127.0.0.1:2181(CONNECTED) 10] ls /brokers/topics                  
[test, lyc.gd4, zerg.hydra]
[zk: 127.0.0.1:2181(CONNECTED) 11] rmr /brokers/topics/zerg.hydra   
[zk: 127.0.0.1:2181(CONNECTED) 12] ls /brokers/topics            
[test, lyc.gd4]
[zk: 127.0.0.1:2181(CONNECTED) 13] 
[zk: 127.0.0.1:2181(CONNECTED) 13] ls /
[consumers, config, controller, admin, brokers, zookeeper, controller_epoch]

[zk: 127.0.0.1:2181(CONNECTED) 15] ls /config/topics
[test, lyc.gd4, zerg.hydra]
[zk: 127.0.0.1:2181(CONNECTED) 16] ls /config/topics/zerg.hydra
[]
[zk: 127.0.0.1:2181(CONNECTED) 17] ls /config/changes
[]
[zk: 127.0.0.1:2181(CONNECTED) 18] ls /config/topics           
[zerg.h, test, lyc.gd4, zerg.hydra]
[zk: 127.0.0.1:2181(CONNECTED) 19] ls /
[consumers, config, controller, admin, brokers, zookeeper, controller_epoch]
[zk: 127.0.0.1:2181(CONNECTED) 21] ls /brokers/topics          
[zerg.h, test, lyc.gd4]
[zk: 127.0.0.1:2181(CONNECTED) 22] ls /brokers/topics
[zerg.h, test, lyc.gd4]
[zk: 127.0.0.1:2181(CONNECTED) 23] ls /brokers/topics
[zerg.h, test, lyc.gd4]
[zk: 127.0.0.1:2181(CONNECTED) 24] 


[root@cdh2 kafka_2.9.2-0.8.1.1]# pwd
/user/local/kafka_2.9.2-0.8.1.1
[root@cdh2 kafka_2.9.2-0.8.1.1]# ls -l
total 408
drwxr-xr-x 3 root root   4096 Jun 28 15:57 bin
drwxr-xr-x 2 root root   4096 Jun 29 00:44 config
drwxr-xr-x 2 root root   4096 Apr 22  2014 libs
-rw-rw-r-- 1 root root  11358 Apr 22  2014 LICENSE
-rw-r--r-- 1 root root      0 Jun 28 15:58 log-cleaner.log
drwxr-xr-x 2 root root   4096 Jun 29 02:00 logs
-rw------- 1 root root 377417 Jun 29 02:51 nohup.out
-rw-rw-r-- 1 root root    162 Apr 22  2014 NOTICE
[root@cdh2 kafka_2.9.2-0.8.1.1]# echo 123,456,789>>test.log
[root@cdh2 kafka_2.9.2-0.8.1.1]# ls -l
total 412
drwxr-xr-x 3 root root   4096 Jun 28 15:57 bin
drwxr-xr-x 2 root root   4096 Jun 29 00:44 config
drwxr-xr-x 2 root root   4096 Apr 22  2014 libs
-rw-rw-r-- 1 root root  11358 Apr 22  2014 LICENSE
-rw-r--r-- 1 root root      0 Jun 28 15:58 log-cleaner.log
drwxr-xr-x 2 root root   4096 Jun 29 02:00 logs
-rw------- 1 root root 377417 Jun 29 02:51 nohup.out
-rw-rw-r-- 1 root root    162 Apr 22  2014 NOTICE
-rw-r--r-- 1 root root     12 Jun 29 03:24 test.log
[root@cdh2 kafka_2.9.2-0.8.1.1]# pwd
/user/local/kafka_2.9.2-0.8.1.1
[root@cdh2 kafka_2.9.2-0.8.1.1]# echo 123,456,789>>test.log
[root@cdh2 kafka_2.9.2-0.8.1.1]# echo 123,456,>>test.log
[root@cdh2 kafka_2.9.2-0.8.1.1]# 


producer
[root@cdh2 kafka_2.9.2-0.8.1.1]# tail -n 0 -f  /user/local/kafka_2.9.2-0.8.1.1/test.log | bin/kafka-console-producer.sh --broker-list localhost:9092,localhost:9093,localhost:9094 --sync --topic zerg.h
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
sdfsdfds###producer 直接输入，不起作用 

1)consumer
[root@cdh2 kafka_2.9.2-0.8.1.1]#  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

2)consumer
[root@cdh2 kafka_2.9.2-0.8.1.1]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

[root@cdh2 kafka_2.9.2-0.8.1.1]# jps
3509 ZooKeeperMain
5759 Jps
2782 Kafka
5585 ConsoleConsumer
2833 Kafka
2813 Kafka
2879 QuorumPeerMain
5701 ConsoleConsumer
5627 ConsoleProducer
[root@cdh2 kafka_2.9.2-0.8.1.1]# 

[root@cdh2 kafka_2.9.2-0.8.1.1]# echo 123,456,789>>test.log
1)
[root@cdh2 kafka_2.9.2-0.8.1.1]#  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
123,456,789
2)
[root@cdh2 kafka_2.9.2-0.8.1.1]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
123,456,789

[root@cdh2 kafka_2.9.2-0.8.1.1]# echo 123,456,789>>test.log
1)
[root@cdh2 kafka_2.9.2-0.8.1.1]#  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
123,456,789
123,456,789
2)
[root@cdh2 kafka_2.9.2-0.8.1.1]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
123,456,789
123,456,789

[root@cdh2 kafka_2.9.2-0.8.1.1]# echo 123,456,>>test.log
1)
[root@cdh2 kafka_2.9.2-0.8.1.1]#  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
123,456,789
123,456,789
123,456,
2)
[root@cdh2 kafka_2.9.2-0.8.1.1]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic zerg.h --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
123,456,789
123,456,789
123,456,

Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一）概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...