Python通过SSH隧道链接Kafka

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python通过SSH隧道链接Kafka相关的知识,希望对你有一定的参考价值。

Python通过SSH隧道链接Kafka

最近有一个需求需要连接Kafka,但是它只允许内网链接,但是有些服务跑在服务器上总没有在我本机调试起来爽,毕竟很多开发工具还是在客户端机器上用的熟练。于是我想到了通过SSH连接Kafka,至于怎么连接可以通过XShellProxifier等等,由于个人还是觉得自己写更灵活,所以我是用Python里的sshtunnel写的(有需要后面我也可以分享下),个人喜好啊,你们自行选择。

由于笔者这里的Kafka环境使用Zookeeper做分布式部署,有多个brokerIP地址,不过这样难不倒,大不了映射多几个端口,一切都似乎很顺利,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(如果没有修改的话),所以即使使用hostsIP解析到本地,但是端口并没有使用配置好的端口。
===============================================================

解决方案:

终于到解决方案了,不啰嗦,直接上。

找到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.pyconn.py导入配置文件

from .self_config import self_design

并且定位到源代码中get_ip_port_afi方法处(该方法将host处理返回ipportafi

# ----------------------------------------

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 处为自己添加的代码

再次连接,成功解决问题

 

 

 

 

 


本文出自 “云驿站 -Leyex学习笔记” 博客,请务必保留此出处http://leyex.blog.51cto.com/4230949/1958453

以上是关于Python通过SSH隧道链接Kafka的主要内容,如果未能解决你的问题,请参考以下文章

使用 Python 通过 SSH 隧道连接到远程 PostgreSQL 数据库

通过vscode的SSH隧道打开一个Matplotlib图

通过 ssh 隧道访问远程数据库(Python 3)

使 Python 通过 SSH 隧道连接到 MySQL

使用 Python Paramiko 通过双 SSH 隧道连接到数据库

ssh隧道 学习总结