如何在java中创建kafka zkstringserializer？

kx7yvsdv 于 2021-06-08 发布在 Kafka

关注(0)|答案(1)|浏览(281)

在搜索如何通过api创建Kafka主题时，我在scala中找到了以下示例：

import kafka.admin.AdminUtils
import kafka.utils.ZKStringSerializer
import org.I0Itec.zkclient.ZkClient

// Create a ZooKeeper client
val sessionTimeoutMs = 10000
val connectionTimeoutMs = 10000
val zkClient = new ZkClient("zookeeper1:2181", sessionTimeoutMs,
                            connectionTimeoutMs, ZKStringSerializer)

// Create a topic with 8 partitions and a replication factor of 3
val topicName = "myTopic"
val numPartitions = 8
val replicationFactor = 3
val topicConfig = new Properties
AdminUtils.createTopic(zkClient, topicName, 
                       numPartitions, replicationFactor, topicConfig)

资料来源：https://stackoverflow.com/a/23360100/871012
最后一个参数 ZKStringSerializer 显然是一个scala物体。我不清楚如何使这个例子在java中工作。
这篇关于如何在clojure中创建scala对象的帖子在clojure中提出了同样的问题，答案是：

ZKStringSerializer$/MODULE$

在java中（我认为）可以转换为：

ZKStringSerializer$.MODULE$

但当我尝试这个（或其他任何数量的变体）时，它们都无法编译。
编译错误为：

KafkaTopicCreator.java:[16,18] cannot find symbol
symbol:   variable ZKStringSerializer$
location: class org.sample.KafkaTopicCreator

我使用的是kafkaè2.9.2-0.8.1.1和java8。

Java scala apache-kafka

来源：https://stackoverflow.com/questions/27006156/how-create-kafka-zkstringserializer-in-java

1条答案

按热度按时间

4urapxun1#

对于java，请尝试以下操作：，
下面的第一个导入语句

import kafka.utils.ZKStringSerializer$;

按以下方式为zkclient创建对象，

String zkHosts = "127.0.0.1:2181"; //If more than one zookeeper then "127.0.0.1:2181,127.0.0.2:2181"
ZkClient zkClient = new ZkClient(zkHosts, 10000, 10000, ZKStringSerializer$.MODULE$);
AdminUtils.createTopic(zkClient, myTopic, 10, 1, new Properties());

上面的代码不适用于kafka>0.9，因为api已经更改，请使用下面的kafka>0.9代码

import java.util.Properties;
import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

public class KafkaTopicCreationInJava
{
    public static void main(String[] args) throws Exception {
        ZkClient zkClient = null;
        ZkUtils zkUtils = null;
        try {
            String zookeeperHosts = "192.168.20.1:2181"; // If multiple zookeeper then -> String zookeeperHosts = "192.168.20.1:2181,192.168.20.2:2181";
            int sessionTimeOutInMs = 15 * 1000; // 15 secs
            int connectionTimeOutInMs = 10 * 1000; // 10 secs

            zkClient = new ZkClient(zookeeperHosts, sessionTimeOutInMs, connectionTimeOutInMs, ZKStringSerializer$.MODULE$);
            zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperHosts), false);

            String topicName = "testTopic";
            int noOfPartitions = 2;
            int noOfReplication = 3;
            Properties topicConfiguration = new Properties();

            AdminUtils.createTopic(zkUtils, topicName, noOfPartitions, noOfReplication, topicConfiguration);

        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (zkClient != null) {
                zkClient.close();
            }
        }
    }
}

赞(0）回复(0）举报 2021-06-08

我来回答

如何在java中创建kafka zkstringserializer？

1条答案

相关问题

热门标签

最新问答