关于消息的发布与订阅，之前一直使用的是activeMQ基于JMS的消息队列进行操作的，最近听说有一个更高效的消息的发布与订阅技术，就是Kafka。

关于kafka的介绍，在这里就不做过多讲解了，因为我自己也不是很了解，大概就知道它与activeMQ一样，都是具有生产者和消费者的发布与订阅消息的机制。

今天我想说的就是，初遇kafka所踩的坑，非常大的坑！！

今天第一次学习Kafka，参考的是ORCHome网上的资料。

具体使用，我这里不过多介绍，具体讲我遇到的问题。因为是自学，我采用的是在centOS6.5的虚拟机上安装的Kafka，由于新版的Kafka自带有zookeeper，所以就直接使用了。

当我按照教程启动玩Kafka后，并且在虚拟机服务器里面是可以正常操作，可是使用JavaAPI远程进行操作的时候，便一直报连接异常！

Java代码：

packagesite.wangxin520.kafkatest;importjava.util.Properties;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.Producer;importorg.apache.kafka.clients.producer.ProducerRecord;public classProducerTest {public static voidmain(String[] args) {

Properties props= newProperties();

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 producer = new KafkaProducer(props);//生产者的主要方法

producer.send(new ProducerRecord("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)

解决办法

这里需要注意的是，因为是远程连接服务器，所以要看服务器的防火墙是否针对端口9092(默认端口)打开的，刚开始弄了很长时间，我一直没弄好的原因是因为中午我重启了服务器，导致防火墙又打开了。

如果防火墙是正常的，就需要改变Kafka的配置：在/config/service.properties中，添加上一句host.name=192.168.211.129

这主要是因为，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客户端发送的消息。

解决成功！