Python通过SSH隧道链接Kafka
最近有一个需求需要连接Kafka，但是它只允许内网链接，但是有些服务跑在服务器上总没有在我本机调试起来爽，毕竟很多开发工具还是在客户端机器上用的熟练。于是我想到了通过SSH连接Kafka，至于怎么连接可以通过XShell、Proxifier等等，由于个人还是觉得自己写更灵活，所以我是用Python里的sshtunnel写的（有需要后面我也可以分享下），个人喜好啊，你们自行选择。
由于笔者这里的Kafka环境使用Zookeeper做分布式部署，有多个broker的IP地址，不过这样难不倒，大不了映射多几个端口，一切都似乎很顺利，SSH隧道建立起来，端口都配置好。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
from
kafka
import
KafkaProducer
import
logging
logging.basicConfig(level
=
logging.INFO)
# 已经配置好的ssh隧道
kafka_host
=
[
'127.0.0.1:19091'
,
'127.0.0.1:19092'
,
'127.0.0.1:19093'
,
]
producer
=
KafkaProducer(bootstrap_servers
=
kafka_host)
producer.send(
'test'
, b
'some_message_bytes'
)
|
开始启动脚本发现问题来了，报错：
|
1
2
3
4
5
6
7
8
9
10
|
Traceback (most recent call last):
File
"<stdin>"
, line
12
,
in
<module>
File
"c:\python27\lib\site-packages\kafka\producer\kafka.py"
, line
347
,
in
__init__
*
*
self
.config)
File
"c:\python27\lib\site-packages\kafka\client_async.py"
, line
220
,
in
__init__
if
self
.config[
'api_version'
]
is
None
:
File
"c:\python27\lib\site-packages\kafka\client_async.py"
, line
861
,
in
check_version
except
Errors.NodeNotReadyError:
NoBrokersAvailable: NoBrokersAvailable
NoBrokersAvailable
|
没找到可连接的Brokers，在服务器上相同的代码却可以直接连接上去了...
于是开始排查，发现Kafka连接时即使使用IP地址配置，但依旧会使用主机名解析IP地址，那么就配置hosts把服务器主机名都指向本地地址（127.0.0.1）不就行了，按道理是可以的，不过发现连接是没问题了，可是消息并没有发送成功。
仔细查看日志，发现：
|
1
|
INFO:kafka.conn:<BrokerConnection node_id
=
258
host
=
kafka
-
04
/
127.0
.
0.1
port
=
9092
>: connecting to
127.0
.
0.1
:
9092
|
确实是把对应的主机名解析到本地地址，但是端口却并没有做相应的改变...
笔者并不是一个轻易放弃的人，于是开始读pykafka的源代码，实际也不是很难，立刻定位到问题了。基本就是pykafka会通过配置好的地址去连接Kafka服务器，并且获取一个可用的地址（这里返回了Kafka的主机名），然后端口会使用默认的Port（如果没有修改的话），所以即使使用hosts把IP解析到本地，但是端口并没有使用配置好的端口。
===============================================================
解决方案：
终于到解决方案了，不啰嗦，直接上。
找到pykafka目录，笔者这里是：
|
1
|
C:\Python27\Lib\site
-
packages\kafka
|
增加一个自定义的配置文件self_config.py（自行修改，这里都是举例）
|
1
2
3
4
5
6
|
self_design
=
{
"kafka-04"
:
19094
,
"kafka-03"
:
19093
,
"kafka-02"
:
19092
,
"kafka-01"
:
19091
,
}
|
分别对client_async.py和conn.py导入配置文件
|
1
|
from
.self_config
import
self_design
|
并且定位到源代码中get_ip_port_afi方法处（该方法将host处理返回ip、port、afi）
# ----------------------------------------
client_async.py
|
1
2
3
4
5
|
host, port, afi
=
get_ip_port_afi(broker.host)
# ==================================
# Self Addon
conn_port
=
port
if
broker.host
in
self_design
else
broker.port
# ==================================
|
并且把以下broker.port修改为conn_port
|
1
2
3
4
|
conn
=
BrokerConnection(host, conn_port, afi,
# broker.port
state_change_callback
=
cb,
node_id
=
node_id,
*
*
self
.config)
|
# ----------------------------------------
conn.py
|
1
2
3
4
5
6
7
8
9
|
if
':'
not
in
host_and_port_str:
# ==================================
# Self Addon
if
host_and_port_str
in
self_design:
af
=
_address_family(host_and_port_str)
return
u
'127.0.0.1'
, self_design[host_and_port_str], af
# ==================================
af
=
_address_family(host_and_port_str)
return
host_and_port_str, DEFAULT_KAFKA_PORT, af
|
# Self Addon 处为自己添加的代码
再次连接，成功解决问题
本文转自 leyex 51CTO博客，原文链接:http://blog.51cto.com/leyex/1958453
