Python通过SSH隧道链接Kafka

CHSHJ

Python通过SSH隧道链接Kafka
　　最近有一个需求需要连接Kafka，但是它只允许内网链接，但是有些服务跑在服务器上总没有在我本机调试起来爽，毕竟很多开发工具还是在客户端机器上用的熟练。于是我想到了通过SSH连接Kafka，至于怎么连接可以通过XShell、Proxifier等等，由于个人还是觉得自己写更灵活，所以我是用Python里的sshtunnel写的（有需要后面我也可以分享下），个人喜好啊，你们自行选择。
　　由于笔者这里的Kafka环境使用Zookeeper做分布式部署，有多个broker的IP地址，不过这样难不倒，大不了映射多几个端口，一切都似乎很顺利，SSH隧道建立起来，端口都配置好。
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')
　　开始启动脚本发现问题来了，报错：
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）不就行了，按道理是可以的，不过发现连接是没问题了，可是消息并没有发送成功。
　　仔细查看日志，发现：
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目录，笔者这里是：
C:\Python27\Lib\site-packages\kafka　　增加一个自定义的配置文件self_config.py（自行修改，这里都是举例）
self_design = {　　
    "kafka-04": 19094,
　　
    "kafka-03": 19093,
　　
    "kafka-02": 19092,
　　
    "kafka-01": 19091,
　　
}
　　分别对client_async.py和conn.py导入配置文件
from .self_config import self_design　　并且定位到源代码中get_ip_port_afi方法处（该方法将host处理返回ip、port、afi）
　　# ----------------------------------------
　　client_async.py
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
conn = BrokerConnection(host, conn_port, afi,  # broker.port　　
                        state_change_callback=cb,
　　
                        node_id=node_id,
　　
                        **self.config)
　　# ----------------------------------------
　　conn.py
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 处为自己添加的代码
　　再次连接，成功解决问题