该死的Kafka,远程连接Kafka超时以及解决办法

关于消息的发布与订阅,之前一直使用的是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");
        //如果请求失败,生产者也会自动重试,即使设置成0 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)

解决办法

  • 这里需要注意的是,因为是远程连接服务器,所以要看服务器的防火墙是否针对端口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客户端发送的消息。

技术分享

解决成功!

文章来自:http://www.cnblogs.com/wangxinblog/p/7623419.html
© 2021 jiaocheng.bubufx.com  联系我们
ICP备案:鲁ICP备09046678号-3