Kafka的大名久仰已久，但一直未曾使用过。入职名客来四个月后公司启动IM项目，IM接口采用的是网易云信的接口。由于网易不提供永久保存历史聊天记录的服务，于是决定将记录保存在公司服务器上。这样通过网易的消息抄送服务，就可以接收网易推送过来的聊天记录信息。考虑到接口的压力会比较大，同时，历史消息不需要高时效性，无须做即时持久化操作，所以可以通过消息队列去处理网易推送过来的抄送消息。在对比了ActiveMQ、RabbitMQ、ZeroMQ、Kafka等常用的消息队列之后，最终决定使用Kafka(主要是考虑后期项目也会用到此利器，因此就斗胆拿IM试刀)。
根据官方文档搭建环境、创建topic、发送消息、消费消息、创建集群都很顺利，下一步就是编写Java代码。从GitHub上下载example，代码都比较简单易懂。一路下来都比较顺利，可就是在发布消息的时候出了问题，调用了producer的send方法后本地接口返回很慢，感觉很不对劲，开始以为是producer的close方法的原因，于是注释掉
//producer.close()；
再次请求，依然很慢。于是添加了回调函数，并打印信息
ProducerManager manager = new ProducerManager();
Producer p = manager.getProducer();
p.send(new ProducerRecord("cc", json.toString()),new Callback() {
@Override
public void onCompletion(RecordMetadata rm, Exception e) {
if(e != null){
e.printStackTrace();
}
}
}
可喜可贺，回调函数抛出异常了
Expiring 1 record(s) for cc-0: 30036 ms has passed since batch creation plus linger time
百度了一下原因，竟然是因为Listeners的配置问题
"PLAINTEXT"表示协议，可选的值有PLAINTEXT和SSL，hostname可以指定IP地址，也可以用"0.0.0.0"表示对所有的网络接口有效，如果hostname为空表示只对默认的网络接口有效,也就是说如果你没有配置advertised.listeners，就使用listeners的配置通告给消息的生产者和消费者，这个过程是在生产者和消费者获取源数据(metadata)。如果都没配置，那么就使用java.net.InetAddress.getCanonicalHostName()返回的值，对于ipv4,基本就是localhost了。然后生产者就会一直卡死，没有反应，原因是你的客户端连接的是127.0.0.1:9092，而不是你期望的服务器的地址。因此只需要把listeners的地址修改成你的服务器地址即可
网友评论