kafka忽略集群Node信息,直接向`bootstrap.servers`地址发送消息-Toy模板网

这篇具有很好参考价值的文章主要介绍了kafka忽略集群Node信息,直接向`bootstrap.servers`地址发送消息。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

背景

当kafka单机安装的时候或者集群安装的时候，kafka会先通过bootstrap.servers获取集群节点。
有时候网络复杂的时候

如内网外部署
地址映射
代理转发等

bootstrap.servers配置地址可能为一个公网地址181.39.77.53:9092,然而返回的节点为内网地址172.16.31.33:9092，此时由于未开通172.16.31.33:9092网络，导致访问失败。

此时通常有两种解决方案文章来源地址https://www.toymoban.com/news/detail-490989.html

开通kafka返回节点的网络
修改kafka连接方式，忽略返回的node节点信息，直接访问bootstrap.servers配置的地址
这里主要介绍下方案二，直接访问bootstrap.servers配置的地址

实现原理

利用类加载机制，重复的类只会加载一次。
同名类先加载的类先生效，后加载的类被忽略
优先加载运行jar包内类。
可以通过在项目内新建同名类（包名也要相同），修改源码覆盖的方式来实现。

实现代码

在项目内新建org.apache.kafka.clients.NetworkClient类
通过修改initiateConnect(Node node, long now)实现

  /**
     * Initiate a connection to the given node
     * @param node the node to connect to
     * @param now current time in epoch milliseconds
     */
    private void initiateConnect(Node node, long now) {

        String nodeConnectionId = node.idString();

        try {
            connectionStates.connecting(nodeConnectionId, now, node.host());
            InetAddress address = connectionStates.currentAddress(nodeConnectionId);

            // 开启单机版kafka连接
            // 如果单机kafka 直接连接`bootstrap.servers`中配置的节点
            if (HookConnectProperties.hooked) {
                KafkaChannel channel = ((Selector) selector).channel("-1");

                // 如果channel为空说明访问的`bootstrap.servers`
                // 不为空则开始访问kafka返回的集群节点
                if(channel != null) {
                    InetSocketAddress remoteAddress = null;
                    try {
                        // 尝试通过反射方式获取`remoteAddress`
                        Field field = KafkaChannel.class.getDeclaredField("remoteAddress");
                        field.setAccessible(true);
                        remoteAddress = (InetSocketAddress)field.get(channel);
                        log.debug("Initiating connection to node {} using address {}", node, address);
                        selector.connect(nodeConnectionId,
                                remoteAddress,
                                this.socketSendBuffer,
                                this.socketReceiveBuffer);

                    } catch (NoSuchFieldException | IllegalAccessException e) {
                        // 获取不到则获取远程地址
                        address = channel.socketAddress();
                        log.debug("Initiating connection to node {} using address {}", node, address);
                        selector.connect(nodeConnectionId,
                                new InetSocketAddress(address, node.port()),
                                this.socketSendBuffer,
                                this.socketReceiveBuffer);
                    }
                    return;
                }
            }
            log.debug("Initiating connection to node {} using address {}", node, address);
            selector.connect(nodeConnectionId,
                    new InetSocketAddress(address, node.port()),
                    this.socketSendBuffer,
                    this.socketReceiveBuffer);
        } catch (IOException e) {
            log.warn("Error connecting to node {}", node, e);
            // Attempt failed, we'll try again after the backoff
            connectionStates.disconnected(nodeConnectionId, now);
            // Notify metadata updater of the connection failure
            metadataUpdater.handleServerDisconnect(now, nodeConnectionId, Optional.empty());
        }
    }

到了这里，关于kafka忽略集群Node信息,直接向`bootstrap.servers`地址发送消息的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！