kafka发送消息的时候报超时，有人遇到过吗？
kafka发送消息的时候报超时，有人遇到过吗？
生产者用的是这个例子：/6
Exception in thread "main" org.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to zookeeper server within timeout: 6000
&at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:880)
&at org.I0Itec.zkclient.ZkClient.&init&(ZkClient.java:98)
&at org.I0Itec.zkclient.ZkClient.&init&(ZkClient.java:84)
&at kafka.consumer.ZookeeperConsumerConnector.connectZk(ZookeeperConsumerConnector.scala:156)
&at kafka.consumer.ZookeeperConsumerConnector.&init&(ZookeeperConsumerConnector.scala:114)
看上去就是你自己网络问题。
--- 共有 1 条评论 ---
可以telnet是通的呀。
你是单服务器，有安装zookeeper吗？？
如果没有，最好修改下server.properties的和zookeeper服务连接地址，直接写为本机地址或者127.0.0.1
--- 共有 1 条评论 ---
是的，找到问题了，机器互相访问的时候，kafka集群要绑定iplogstash联合python kafka进行异常数据监控报警 - 推酷
logstash联合python kafka进行异常数据监控报警
kafka的作用，我想大家应该知道…. 一个可以支撑大流量的消息系统…
懒人是有懒人的想法和做法的.. &为毛这么说？ 现在所有的日志都通过logstash达到了elasticsearch里面。 我们可以通过kibana来进行查看，但是问题是，我们如何针对某些异常进行告警… & 比如我临时要过滤些条件，比如xx.com是否出现了大量的超时，还有就是针对200的http code的行为，进行metric。某些页面因为改版，导致我们抓不到他的具体信息了。
网站的原文是 &
& ，妈蛋的 &总是被抓取….
其实早已经分析出他们的ip地址，只是自己懒得理罢了。。。
对于牛逼的kafka服务端的安装，是有些麻烦的，另外它是需要zookeeper协同调度的。 & 我这边就不说kafka的安装的整个流程了，大家自己找个文档搞搞就行了。&我们的模块都是在docker服务器上，就顺手也pull了一个docker化的kafka和zookeeper。&
docker pull spotify/kafka
docker run -p
--env ADVERTISED_HOST=`boot2docker ip` --env ADVERTISED_PORT=9092 spotify/kafka
export KAFKA=`boot2docker ip`:9092
kafka-console-producer.sh --broker-list $KAFKA --topic test
export ZOOKEEPER=`boot2docker ip`:2181
kafka-console-consumer.sh --zookeeper $ZOOKEEPER --topic test
下面的是logstash相关的配置，我们以前是直接在agent端把output扔到es里面，随着量级的增大，就改放到了redis里做broker消息队列。 &现在要利用kafka做发布，那么我们只需要加入kafka就可以了。&
这里有人可能疑惑了，为毛不直接解析进行监控？ &直接Elasticsearch？ 你怎么知道什么时候会出现某个条件的异常？ &难道要实时的去query es？ & 大家要知道logstash很好的帮助大家做好了日志的切分，tail，offset机制。
zk_connect =& … # string (optional), default: “localhost:2181″
group_id =& … # string (optional), default: “logstash”
topic_id =& … # string (optional), default: “test”
reset_beginning =& … # boolean (optional), default: false
consumer_threads =& … # number (optional), default: 1
queue_size =& … # number (optional), default: 20
rebalance_max_retries =& … # number (optional), default: 4
rebalance_backoff_ms =& … # number (optional), default: &2000
consumer_timeout_ms =& … # number (optional), default: -1
consumer_restart_on_error =& … # boolean (optional), default: true
consumer_restart_sleep_ms =& … # number (optional), default: 0
decorate_events =& … # boolean (optional), default: false
consumer_id =& … # string (optional) default: nil
fetch_message_max_bytes =& … # number (optional) default: 1048576
这是output输出
# output {
broker_list =& … # string (optional), default: “localhost:9092″
topic_id =& … # string (optional), default: “test”
compression_codec =& … # string (optional), one of [&none&, &gzip&, &snappy&], default: “none”
compressed_topics =& … # string (optional), default: “”
request_required_acks =& … # number (optional), one of [-1, 0, 1], default: 0
serializer_class =& … # string, (optional) default: “kafka.serializer.StringEncoder”
partitioner_class =& … # string (optional) default: “kafka.producer.DefaultPartitioner”
request_timeout_ms =& … # number (optional) default: 10000
producer_type =& … # string (optional), one of [&sync&, &async&] default =& ‘sync’
key_serializer_class =& … # string (optional) default: “kafka.serializer.StringEncoder”
message_send_max_retries =& … # number (optional) default: 3
retry_backoff_ms =& … # number (optional) default: 100
topic_metadata_refresh_interval_ms =& … # number (optional) default: 600 * 1000
queue_buffering_max_ms =& … # number (optional) default: 5000
queue_buffering_max_messages =& … # number (optional) default: 10000
queue_enqueue_timeout_ms =& … # number (optional) default: -1
batch_num_messages =& … # number (optional) default: 200
send_buffer_bytes =& … # number (optional) default: 100 * 1024
client_id =& … # string (optional) default: “”
上线是关于logstash里面关于kafka的input output的配置，我们再来看看同时订阅kafka broker的python程序是如何跑的。 & 具体代码我就不贴了，你可以通过一个demo很容易进行扩展.&
#!/usr/bin/env python
import threading, logging, time
from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer
from kafka.producer import SimpleProducer
#这个是喂数据的
class Producer(threading.Thread):
daemon = True
def run(self):
client = KafkaClient(&localhost:9092&)
producer = SimpleProducer(client)
while True:
producer.send_messages('my-topic', &test&)
producer.send_messages('my-topic', &\xc2Hola, mundo!&)
time.sleep(1)
#这个是消费者，也就是咱们从kafka取数据的逻辑。
class Consumer(threading.Thread):
daemon = True
def run(self):
client = KafkaClient(&localhost:9092&)
consumer = SimpleConsumer(client, &test-group&, &my-topic&)
for message in consumer:
print(message)
def main():
threads = [
Producer(),
Consumer()
for t in threads:
time.sleep(5)
if __name__ == &__main__&:
logging.basicConfig(
format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:%(levelname)s:%(process)d:%(message)s',
level=logging.DEBUG
另外python kafka里面是支持多个 consumer的。。。。
from kafka import KafkaClient, MultiProcessConsumer
kafka = KafkaClient(&localhost:9092&)
consumer = MultiProcessConsumer(kafka, &my-group&, &my-topic&, num_procs=2)
consumer = MultiProcessConsumer(kafka, &my-group&, &my-topic&,
partitions_per_proc=2)
for message in consumer:
print(message)
for message in consumer.get_messages(count=5, block=True, timeout=4):
print(message)
就先这样了，有时间我会再把这文章扩展下。
已发表评论数()
请填写推刊名
描述不能大于100个字符!
权限设置： 公开
仅自己可见
正文不准确
标题不准确
排版有问题
主题不准确
没有分页内容
图片无法显示
视频无法显示
与原文不一致本类文章总排行版
本类文章本月排行
本类文章本周排行2015年3月 Linux/Unix社区大版内专家分月排行榜第二2014年12月 Linux/Unix社区大版内专家分月排行榜第二2014年11月 Linux/Unix社区大版内专家分月排行榜第二2014年9月 Linux/Unix社区大版内专家分月排行榜第二2014年8月 Linux/Unix社区大版内专家分月排行榜第二2014年7月 Linux/Unix社区大版内专家分月排行榜第二2014年5月 Linux/Unix社区大版内专家分月排行榜第二2014年4月 Linux/Unix社区大版内专家分月排行榜第二
2014年3月 Linux/Unix社区大版内专家分月排行榜第三
2015年3月 Linux/Unix社区大版内专家分月排行榜第二2014年12月 Linux/Unix社区大版内专家分月排行榜第二2014年11月 Linux/Unix社区大版内专家分月排行榜第二2014年9月 Linux/Unix社区大版内专家分月排行榜第二2014年8月 Linux/Unix社区大版内专家分月排行榜第二2014年7月 Linux/Unix社区大版内专家分月排行榜第二2014年5月 Linux/Unix社区大版内专家分月排行榜第二2014年4月 Linux/Unix社区大版内专家分月排行榜第二
2014年3月 Linux/Unix社区大版内专家分月排行榜第三
2015年3月 Linux/Unix社区大版内专家分月排行榜第二2014年12月 Linux/Unix社区大版内专家分月排行榜第二2014年11月 Linux/Unix社区大版内专家分月排行榜第二2014年9月 Linux/Unix社区大版内专家分月排行榜第二2014年8月 Linux/Unix社区大版内专家分月排行榜第二2014年7月 Linux/Unix社区大版内专家分月排行榜第二2014年5月 Linux/Unix社区大版内专家分月排行榜第二2014年4月 Linux/Unix社区大版内专家分月排行榜第二
2014年3月 Linux/Unix社区大版内专家分月排行榜第三
2015年3月 Linux/Unix社区大版内专家分月排行榜第二2014年12月 Linux/Unix社区大版内专家分月排行榜第二2014年11月 Linux/Unix社区大版内专家分月排行榜第二2014年9月 Linux/Unix社区大版内专家分月排行榜第二2014年8月 Linux/Unix社区大版内专家分月排行榜第二2014年7月 Linux/Unix社区大版内专家分月排行榜第二2014年5月 Linux/Unix社区大版内专家分月排行榜第二2014年4月 Linux/Unix社区大版内专家分月排行榜第二
2014年3月 Linux/Unix社区大版内专家分月排行榜第三
匿名用户不能发表回复！|
每天回帖即可获得10分可用分！小技巧：
你还可以输入10000个字符
(Ctrl+Enter)
请遵守CSDN，不得违反国家法律法规。
转载文章请注明出自“CSDN（www.csdn.net）”。如是商业用途请联系原作者。}