search

adyliu / jafka

a fast and simple distributed publish-subscribe messaging system (mq)
https://github.com/adyliu/jafka/wiki
Apache License 2.0
719 stars 323 forks source link

consumer的问题 #54

Open linyuan123 opened 7 years ago

linyuan123 commented 7 years ago

[hadoop@ZK1 bin]$ ./consumer-console.sh --zookeeper 192.168.1.97:2181,192.168.1.96:2181,192.168.1.98:2181 --topic ly --from-beginning log4j:WARN No appenders could be found for logger (com.sohu.jafka.consumer.ZookeeperConsumerConnector). log4j:WARN Please initialize the log4j system properly. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. Exception in thread "main" java.lang.NullPointerException at com.github.zkclient.ZkClient.deleteRecursive(ZkClient.java:442) at com.sohu.jafka.console.ConsoleConsumer.tryCleanupZookeeper(ConsoleConsumer.java:174) at com.sohu.jafka.console.ConsoleConsumer.main(ConsoleConsumer.java:106)

消费者接受消息为什么会报错啊，我用给的java demo 是可以接收的 ，但是这个却不可以。求解

adyliu commented 7 years ago

配置下log4j.properties，在conf下有个样例。将日志输出都贴在下面。

linyuan123 commented 7 years ago

2016-11-03 14:25:39.292 INFO ZookeeperConsumerConnector - Connecting to zookeeper instance at 192.168.1.97:2181,192.168.1.98:2181,192.168.1.96:2181 2016-11-03 14:25:39.312 INFO ZkEventThread - Starting ZkClient event thread. 2016-11-03 14:25:39.381 INFO ClientCnxn$SendThread - Opening socket connection to server ZK2/192.168.1.96:2181. Will not attempt to authenticate using SASL (unknown error) 2016-11-03 14:25:39.383 INFO ClientCnxn$SendThread - Socket connection established to ZK2/192.168.1.96:2181, initiating session 2016-11-03 14:25:39.389 INFO ClientCnxn$SendThread - Session establishment complete on server ZK2/192.168.1.96:2181, sessionid = 0x1582813b9690009, negotiated timeout = 6000 2016-11-03 14:25:39.390 INFO ZkClient - zookeeper state changed (SyncConnected) 2016-11-03 14:25:39.390 INFO ZookeeperConsumerConnector - Connected to zookeeper at 192.168.1.97:2181,192.168.1.98:2181,192.168.1.96:2181 2016-11-03 14:25:39.391 INFO ZookeeperConsumerConnector - starting auto committer every 1000 ms 2016-11-03 14:25:39.477 INFO ZookeeperConsumerConnector - create message stream by consumerid [dev-01-1478154339475-ca6c863c] with groupid [test_group] 2016-11-03 14:25:39.618 INFO ZookeeperConsumerConnector$ZKRebalancerListener - starting watcher executor thread for consumer test_group_dev-01-1478154339475-ca6c863c 2016-11-03 14:25:39.652 INFO ZookeeperConsumerConnector - register consumer in zookeeper [/consumers/test_group/ids/test_group_dev-01-1478154339475-ca6c863c] => [{"ly":1}] 2016-11-03 14:25:39.671 INFO ZookeeperConsumerConnector$ZKRebalancerListener - [test_group_dev-01-1478154339475-ca6c863c] rebalancing starting. try #0 2016-11-03 14:25:39.714 INFO ZookeeperConsumerConnector$ZKRebalancerListener - Committing all offsets after clearing the fetcher queues 2016-11-03 14:25:39.714 INFO ZookeeperConsumerConnector$ZKRebalancerListener - Releasing partition ownership => [] 2016-11-03 14:25:39.716 INFO ZookeeperConsumerConnector$ZKRebalancerListener - Consumer test_group_dev-01-1478154339475-ca6c863c rebalancing the following 1 partitions of topic ly: [1-0] with 1 consumers: [test_group_dev-01-1478154339475-ca6c863c-0] 2016-11-03 14:25:39.716 INFO ZookeeperConsumerConnector$ZKRebalancerListener - [test_group_dev-01-1478154339475-ca6c863c-0] ==> 1-0 claimming 2016-11-03 14:25:41.983 ERROR ZookeeperConsumerConnector$ZKRebalancerListener - error in earliestOrLatestOffset() java.nio.channels.UnresolvedAddressException at sun.nio.ch.Net.checkAddress(Unknown Source) at sun.nio.ch.SocketChannelImpl.connect(Unknown Source) at com.sohu.jafka.network.BlockingChannel.connect(BlockingChannel.java:58) at com.sohu.jafka.consumer.SimpleOperation.connect(SimpleOperation.java:67) at com.sohu.jafka.consumer.SimpleOperation.getOrMakeConnection(SimpleOperation.java:91) at com.sohu.jafka.consumer.SimpleOperation.send(SimpleOperation.java:98) at com.sohu.jafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.java:59) at com.sohu.jafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.earliestOrLatestOffset(ZookeeperConsumerConnector.java:698) at com.sohu.jafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.addPartitionTopicInfo(ZookeeperConsumerConnector.java:659) at com.sohu.jafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.rebalance(ZookeeperConsumerConnector.java:581) at com.sohu.jafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.java:478) at com.sohu.jafka.consumer.ZookeeperConsumerConnector.consume(ZookeeperConsumerConnector.java:259) at com.sohu.jafka.consumer.ZookeeperConsumerConnector.createMessageStreams(ZookeeperConsumerConnector.java:206) at tbkl.main.test.JJja.main(JJja.java:32) 2016-11-03 14:25:41.988 INFO ZookeeperConsumerConnector$ZKRebalancerListener - [test_group_dev-01-1478154339475-ca6c863c] rebalanced OK. try #0, cost 2317 ms 2016-11-03 14:25:41.990 INFO FetcherRunnable - FetchRunnable-1-0 comsume at ZK1:9092 with [ly-1-0,] 2016-11-03 14:25:41.991 ERROR FetcherRunnable - error in FetcherRunnable java.nio.channels.UnresolvedAddressException at sun.nio.ch.Net.checkAddress(Unknown Source) at sun.nio.ch.SocketChannelImpl.connect(Unknown Source) at com.sohu.jafka.network.BlockingChannel.connect(BlockingChannel.java:58) at com.sohu.jafka.consumer.SimpleOperation.connect(SimpleOperation.java:67) at com.sohu.jafka.consumer.SimpleOperation.getOrMakeConnection(SimpleOperation.java:91) at com.sohu.jafka.consumer.SimpleOperation.send(SimpleOperation.java:98) at com.sohu.jafka.consumer.SimpleConsumer.multifetch(SimpleConsumer.java:64) at com.sohu.jafka.consumer.FetcherRunnable.fetchOnce(FetcherRunnable.java:137) at com.sohu.jafka.consumer.FetcherRunnable.run(FetcherRunnable.java:104)

linyuan123 commented 7 years ago

还有用给的java demo 发送消息也报错了 2016-11-03 14:26:47.554 INFO ZkEventThread - Starting ZkClient event thread. 2016-11-03 14:26:47.626 INFO ClientCnxn$SendThread - Opening socket connection to server ZK2/192.168.1.96:2181. Will not attempt to authenticate using SASL (unknown error) 2016-11-03 14:26:47.628 INFO ClientCnxn$SendThread - Socket connection established to ZK2/192.168.1.96:2181, initiating session 2016-11-03 14:26:47.636 INFO ClientCnxn$SendThread - Session establishment complete on server ZK2/192.168.1.96:2181, sessionid = 0x1582813b969000a, negotiated timeout = 6000 2016-11-03 14:26:47.638 INFO ZkClient - zookeeper state changed (SyncConnected) 2016-11-03 14:26:47.643 INFO ZKBrokerPartitionInfo - read all brokers count: 1 2016-11-03 14:26:47.648 INFO ZKBrokerPartitionInfo - Loading Broker ZK1-1478153990449:ZK1:9092:true 2016-11-03 14:26:47.667 INFO ProducerPool - Creating sync producer for broker id = 1 at ZK1:9092 2016-11-03 14:26:49.924 INFO ProducerPool - Closing all sync producers 2016-11-03 14:26:49.925 INFO ZkEventThread - Terminate ZkClient event thread. Exception in thread "main" 2016-11-03 14:26:49.928 INFO ClientCnxn$EventThread - EventThread shut down java.nio.channels.UnresolvedAddressException at sun.nio.ch.Net.checkAddress(Unknown Source) at sun.nio.ch.SocketChannelImpl.connect(Unknown Source) at com.sohu.jafka.network.BlockingChannel.connect(BlockingChannel.java:58) at com.sohu.jafka.producer.SyncProducer.connect(SyncProducer.java:103) at com.sohu.jafka.producer.SyncProducer.send(SyncProducer.java:85) at com.sohu.jafka.producer.SyncProducer.send(SyncProducer.java:77) at com.sohu.jafka.producer.ProducerPool.send(ProducerPool.java:165) at com.sohu.jafka.producer.Producer.zkSend(Producer.java:207) at com.sohu.jafka.producer.Producer.send(Producer.java:171) at tbkl.main.test.JJProducer.main(JJProducer.java:44)

adyliu commented 7 years ago

java.nio.channels.UnresolvedAddressException --> ZK1:9092

无法解析ZK1的域名，解决方法（任选一种）：

  1. 绑定HOST (ZK1...)
  2. 修改jafka/conf/server.properties中的hostname参数为本机的IP地址
linyuan123 commented 7 years ago

hosts 文件下我已经配置了 192.168.1.97 ZK1 192.168.1.96 ZK2 192.168.1.98 ZK3

还是一样的错误

adyliu commented 7 years ago

需要在 producer/consumer出设置，目前是 producer和consumer服务解析hostname

linyuan123 commented 7 years ago

但是我第一篇安装的时候并没有报错啊server.properties中的hostname参数为本机的IP地址 我都没有设置

adyliu commented 7 years ago

如果是在一台机器上执行的话，hosts文件会解析的，跨机器如果没有域名就需要ip地址

linyuan123 commented 7 years ago

不在一台机器上 zookeeper 集群在VMware vSphere 搭建的 。然后本地win8系统 的eclipse 跑的 java demo

adyliu commented 7 years ago

java有DNS缓存，如果eclipse的java程序hosts没有生效，需要重启下eclispe。

另外，如果服务器(broker)没有域名的话，建议使用ip地址，这样每一个producer/consumer都无需设置hosts文件

linyuan123 commented 7 years ago

因为后面还要在这三台虚拟机上搭建hadoop 和hbase 所以hosts 文件就配置了域名解析

linyuan123 commented 7 years ago

eclipse重启之后还是报同样的错误

adyliu commented 7 years ago

eclipse这台机器上ping ZK1