标签：max 百度百科 baidu 高效 一个 远程连接 ges 安装 and
关于消息的发布与订阅，之前一直使用的是activeMQ基于JMS的消息队列进行操作的，最近听说有一个更高效的消息的发布与订阅技术，就是Kafka。
关于kafka的介绍，在这里就不做过多讲解了，因为我自己也不是很了解，大概就知道它与activeMQ一样，都是具有生产者和消费者的发布与订阅消息的机制。
具体请参见百度百科Apache Kafka。
今天我想说的就是，初遇kafka所踩的坑，非常大的坑！！
今天第一次学习Kafka，参考的是ORCHome网上的资料。
具体使用，我这里不过多介绍，具体讲我遇到的问题。因为是自学，我采用的是在centOS6.5的虚拟机上安装的Kafka，由于新版的Kafka自带有zookeeper，所以就直接使用了。
当我按照教程启动玩Kafka后，并且在虚拟机服务器里面是可以正常操作，可是使用JavaAPI远程进行操作的时候，便一直报连接异常！
Java代码：
package site.wangxin520.kafkatest; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; public class ProducerTest { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "http://192.168.211.129:9092"); //The "all" setting we have specified will result in blocking on the full commit of the record, the slowest but most durable setting. //“所有”设置将导致记录的完整提交阻塞，最慢的，但最持久的设置。 props.put("acks", "all"); //如果请求失败，生产者也会自动重试，即使设置成０ the producer can automatically retry. props.put("retries", 0); //The producer maintains buffers of unsent records for each partition. props.put("batch.size", 16384); //默认立即发送，这里这是延时毫秒数 props.put("linger.ms", 1); //生产者缓冲大小，当缓冲区耗尽后，额外的发送调用将被阻塞。时间超过max.block.ms将抛出TimeoutException props.put("buffer.memory", 33554432); //The key.serializer and value.serializer instruct how to turn the key and value objects the user provides with their ProducerRecord into bytes. props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //创建kafka的生产者类 Producer<String, String> producer = new KafkaProducer<String, String>(props); //生产者的主要方法 producer.send(new ProducerRecord<String, String>("show", "测试Kafka")); producer.close(); } }
代码没问题，但是每次运行就会抛一个time out 异常，总是连接失败。
java.net.ConnectException: Connection refused: no further information at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739) at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:51) at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:73) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:323) at org.apache.kafka.common.network.Selector.poll(Selector.java:291) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260) at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:236) at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:148) at java.lang.Thread.run(Thread.java:745)
解决办法
这主要是因为，kafka默认是监听localhost的端口，如果不配置新端口名的话，就解析监听不到消息。
现在重新启动一下，看看是不是已经解决了。
在kafka安装目录，启动自带的zookeeper服务：
bin/zookeeper-server-start.sh config/zookeeper.properties
在同一个地方，启动kafka服务
bin/kafka-server-start.sh config/server.properties
使用消费者客户端，监听show的topic，验证是否已经启动了Kafka
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic show --from-beginning
没有报错，并且现在服务器端已经在监听状态
启动Java客户端，控制台没有报错
并且在服务器端显示了刚刚在Java客户端发送的消息。
解决成功！
标签：max 百度百科 baidu 高效 一个 远程连接 ges 安装 and
原文地址：http://www.cnblogs.com/wangxinblog/p/7623419.html
