Linux安装kafka 分布式
2015-06-16 15:18
656 查看
下载kafka,地址:http://kafka.apache.org/downloads.html选择Binary
downloads下载
上传到linux并解压
修改server.properties文件
启动zookeeper (安装略…..)
在每台机器上启动kafka 启动脚本在bin目录下
创建topic:创建topic:创建topic:
查看Topic:
单机模式收发消息
发送消息:
接收消息:
发送消息和接收消息是在两个窗口
在发送消息窗口输入字符串,在接收消息窗口查看
分布式收发消息
发送消息:
接收消息:
写--from-beginning会显示历史消息,如果只想显示最新的可以不写
停止服务:
报错和解决:
这个错误是config/server.properties的host.name写的不对,可能是前面的“#”没有去掉或是写的主机名称,改成服务器ip地址就可以了,如果改成localhost单机模式不会有问题,但分布式的时候会报下面错误。
downloads下载
上传到linux并解压
$ tar -zxvf kafka_2.11-0.8.2.1.tgz
修改server.properties文件
$ vi config/server.properties
broker.id=1 (每台机器不能一样,我的另外两台机器写的是2,3) host.name=192.168.10.117 (写服务器ip) zookeeper.connect= 192.168.10.117:2181, 192.168.10.114:2181, 192.168.10.115:2181
启动zookeeper (安装略…..)
在每台机器上启动kafka 启动脚本在bin目录下
$ cd kafka_2.11-0.8.2.1/bin/ $ ./kafka-server-start.sh ../config/server.properties &
创建topic:创建topic:创建topic:
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic mykafka
查看Topic:
./kafka-topics.sh --list --zookeeper localhost:2181
单机模式收发消息
发送消息:
./kafka-console-producer.sh --broker-list localhost:9092 --topic mykafka
接收消息:
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic mykafka --from-beginning
发送消息和接收消息是在两个窗口
在发送消息窗口输入字符串,在接收消息窗口查看
分布式收发消息
发送消息:
./kafka-console-producer.sh --broker-list 192.168.10.114:9092 --topic mykafka
接收消息:
./kafka-console-consumer.sh --zookeeper 192.168.10.114:2181 --topic mykafka --from-beginning
写--from-beginning会显示历史消息,如果只想显示最新的可以不写
停止服务:
./kafka-server-stop.sh
报错和解决:
[2015-06-16 11:24:13,015] ERROR Failed to send requests for topics mykafka with correlation ids in [0,8] (kafka.producer.async.DefaultEventHandler) [2015-06-16 11:24:13,015] ERROR Error in handling batch of 1 events (kafka.producer.async.ProducerSendThread) kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries. at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68) at scala.collection.immutable.Stream.foreach(Stream.scala:594) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
这个错误是config/server.properties的host.name写的不对,可能是前面的“#”没有去掉或是写的主机名称,改成服务器ip地址就可以了,如果改成localhost单机模式不会有问题,但分布式的时候会报下面错误。
[2015-06-16 14:20:59,519] WARN Fetching topic metadata with correlation id 9 for topics [Set(mykafka)] from broker [id:0,host:192.168.10.114,port:9092] failed (kafka.client.ClientUtils$) java.nio.channels.ClosedChannelException at kafka.network.BlockingChannel.send(BlockingChannel.scala:100) at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72) at kafka.producer.SyncProducer.send(SyncProducer.scala:113) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58) at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82) at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67) at kafka.utils.Utils$.swallow(Utils.scala:172) at kafka.utils.Logging$class.swallowError(Logging.scala:106) at kafka.utils.Utils$.swallowError(Utils.scala:45) at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68) at scala.collection.immutable.Stream.foreach(Stream.scala:594) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45) [2015-06-16 14:20:59,520] ERROR fetching topic metadata for topics [Set(mykafka)] from broker [ArrayBuffer(id:0,host:192.168.10.114,port:9092)] failed (kafka.utils.Utils$) kafka.common.KafkaException: fetching topic metadata for topics [Set(mykafka)] from broker [ArrayBuffer(id:0,host:192.168.10.114,port:9092)] failed at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:72) at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82) at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67) at kafka.utils.Utils$.swallow(Utils.scala:172) at kafka.utils.Logging$class.swallowError(Logging.scala:106) at kafka.utils.Utils$.swallowError(Utils.scala:45) at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67) at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88) at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68) at scala.collection.immutable.Stream.foreach(Stream.scala:594) at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67) at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45) Caused by: java.nio.channels.ClosedChannelException at kafka.network.BlockingChannel.send(BlockingChannel.scala:100) at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72) at kafka.producer.SyncProducer.send(SyncProducer.scala:113) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58) ... 12 more
相关文章推荐
- linux 命令——28 tar
- Centos 6.6 安装ICE
- Linux基础与命令
- Linux流量监控工具 - iftop (最全面的iftop教程)
- linux 命令——27 chmod
- 一个迷茫的 大学生自学Linux的开始
- linux学习 命令ll后字段的解释(转)
- linux学习 命令ll后字段的解释(转)
- linux 字符集更改
- linux 命令——25 linux文件属性详解
- Linux文件系统
- linux 命令——24 Linux文件类型与扩展名
- gitlab 一键安装 centos 6.5
- linux下的mysql命令
- linux时间同步,ntpd、ntpdate
- linux 命令——23 目录结构
- 如何在 Linux 上用 IP转发使内部网络连接到互联网
- centos 下的 ruby on ralis 配置
- 剖析 Linux hypervisor
- linux内核部件分析(三)——记录生命周期的kref