Docker Kafka 容器消费者不消费数据

Question

我是 Docker 和 Apache Kafka 的新手。我想做的是在java中创建一个消费者和生产者类。我设置了 spotify/kafka，它是 Docker 的 kafka 容器。但是出了点问题。

我找不到任何 docker kafka 容器的生产者消费者示例（如果你有，请分享），所以我只是试着像普通的 kafka 一样做（我的意思不是作为 docker 容器，我想有用法没有区别）。我试过这段代码here（我也试图联系这个人问但无法实现，所以我在这里寻求帮助）：但是当我向生产者终端写一些东西时，生产者终端什么也没有出现。我的操作系统是 Ubuntu Xenial 16.04。这是我所做的：

我通过输入以下内容启动了 docker kafka 容器：

docker run -it spotify/kafka

在输出结束时我收到了这条消息，所以我猜它开始正确：

2018-02-25 09:27:16,911 INFO success: kafka entered RUNNING state, process has stayed up for > than 1 seconds (startsecs)

消费类：

public class Consumer {
private static Scanner in;

public static void main(String[] argv)throws Exception{
    if (argv.length != 2) {
        System.err.printf("Usage: %s <topicName> <groupId>\n",
                Consumer.class.getSimpleName());
        System.exit(-1);
    }
    in = new Scanner(System.in);
    String topicName = argv[0];
    String groupId = argv[1];

    ConsumerThread consumerRunnable = new ConsumerThread(topicName,groupId);
    consumerRunnable.start();

    String line = "";
    while (!line.equals("exit")) {

        line = in.next();
    }
    consumerRunnable.getKafkaConsumer().wakeup();
    System.out.println("Stopping consumer .....");
    consumerRunnable.join();
}

private static class ConsumerThread extends Thread{
    private String topicName;
    private String groupId;
    private KafkaConsumer<String,String> kafkaConsumer;

    public ConsumerThread(String topicName, String groupId){
        this.topicName = topicName;
        this.groupId = groupId;
    }
    public void run() {
        Properties configProperties = new Properties();
        configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        configProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        configProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        configProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, "simple");

        //Figure out where to start processing messages from
        kafkaConsumer = new KafkaConsumer<String, String>(configProperties);
        kafkaConsumer.subscribe(Arrays.asList(topicName));
        //Start processing messages
        try {
            while (1) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);

        System.out.println(records.toString() +"geldi");
                for (ConsumerRecord<String, String> record : records)
                    System.out.println(record.value());
            }
        }catch(WakeupException ex){
            System.out.println("Exception caught " + ex.getMessage());
        }finally{
            kafkaConsumer.close();
            System.out.println("After closing KafkaConsumer");
        }
    }
    public KafkaConsumer<String,String> getKafkaConsumer(){
       return this.kafkaConsumer;
    }
}
}

生产者类：

public class Producer {
private static Scanner in;
public static void main(String[] argv)throws Exception {
    if (argv.length != 1) {
        System.err.println("Please specify 1 parameters ");
        System.exit(-1);
    }
    String topicName = argv[0];
    in = new Scanner(System.in);
    System.out.println("Enter message(type exit to quit)");

    //Configure the Producer
    Properties configProperties = new Properties();
    configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
    configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");
    configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

    org.apache.kafka.clients.producer.Producer producer = new KafkaProducer(configProperties);
    String line = in.nextLine();
    while(!line.equals("exit")) {
        //TODO: Make sure to use the ProducerRecord constructor that does not take parition Id
        ProducerRecord<String, String> rec = new ProducerRecord<String, String>(topicName,line);
        producer.send(rec);
        line = in.nextLine();
    }
    in.close();
    producer.close();
}
}

在不同的终端输入两个类后：

mvn clean compile assembly:single
java -cp (fat jar path) .../Consumer test(topic name) group1
java -cp (fat jar path) .../Producer test(topic name)

当我在生产者终端中输入内容时，消费者中什么也没有出现。请注意，我没有安装 zookeeper，因为 spotify/kafka 包含 zookeeper。在执行这些步骤之前，我没有创建任何主题或组。这些是我唯一做的事情。我找不到如何做到这一点。我该如何解决这个问题？

编辑：我添加了消费者和生产者配置值，有人知道有什么错误吗？

消费者配置：

    metric.reporters = []
metadata.max.age.ms = 300000
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
group.id = gr1
partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]
reconnect.backoff.ms = 50
sasl.kerberos.ticket.renew.window.factor = 0.8
max.partition.fetch.bytes = 1048576
bootstrap.servers = [localhost:9092]
retry.backoff.ms = 100
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
ssl.keystore.type = JKS
ssl.trustmanager.algorithm = PKIX
enable.auto.commit = true
ssl.key.password = null
fetch.max.wait.ms = 500
sasl.kerberos.min.time.before.relogin = 60000
connections.max.idle.ms = 540000
ssl.truststore.password = null
session.timeout.ms = 30000
metrics.num.samples = 2
client.id = simple
ssl.endpoint.identification.algorithm = null
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
ssl.protocol = TLS
check.crcs = true
request.timeout.ms = 40000
ssl.provider = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.keystore.location = null
heartbeat.interval.ms = 3000
auto.commit.interval.ms = 5000
receive.buffer.bytes = 32768
ssl.cipher.suites = null
ssl.truststore.type = JKS
security.protocol = PLAINTEXT
ssl.truststore.location = null
ssl.keystore.password = null
ssl.keymanager.algorithm = SunX509
metrics.sample.window.ms = 30000
fetch.min.bytes = 1024
send.buffer.bytes = 131072
auto.offset.reset = latest

2018-02-25 16:23:37 INFO  AppInfoParser:82 - Kafka version : 0.9.0.0
2018-02-25 16:23:37 INFO  AppInfoParser:83 - Kafka commitId :     fc7243c2af4b2b4a

生产者配置：

compression.type = none
metric.reporters = []
metadata.max.age.ms = 300000
metadata.fetch.timeout.ms = 60000
reconnect.backoff.ms = 50
sasl.kerberos.ticket.renew.window.factor = 0.8
bootstrap.servers = [localhost:9092]
retry.backoff.ms = 100
sasl.kerberos.kinit.cmd = /usr/bin/kinit
buffer.memory = 33554432
timeout.ms = 30000
key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
ssl.keystore.type = JKS
ssl.trustmanager.algorithm = PKIX
block.on.buffer.full = false
ssl.key.password = null
max.block.ms = 60000
sasl.kerberos.min.time.before.relogin = 60000
connections.max.idle.ms = 540000
ssl.truststore.password = null
max.in.flight.requests.per.connection = 5
metrics.num.samples = 2
client.id = 
ssl.endpoint.identification.algorithm = null
ssl.protocol = TLS
request.timeout.ms = 30000
ssl.provider = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
acks = 1
batch.size = 16384
ssl.keystore.location = null
receive.buffer.bytes = 32768
ssl.cipher.suites = null
ssl.truststore.type = JKS
security.protocol = PLAINTEXT
retries = 0
max.request.size = 1048576
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
ssl.truststore.location = null
ssl.keystore.password = null
ssl.keymanager.algorithm = SunX509
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
send.buffer.bytes = 131072
linger.ms = 0

2018-02-25 16:24:16 INFO  AppInfoParser:82 - Kafka version : 0.9.0.0
2018-02-25 16:24:16 INFO  AppInfoParser:83 - Kafka commitId : fc7243c2af4b2b4a

Answer 1

经过长时间的搜索，我发现了问题。当我运行我使用的 docker kafka 容器 ches/kafka 时，我没有指定端口号。

docker run -d -p 2181:2181 --name zookeeper jplock/zookeeper
docker run -d -p 9092 --name kafka --link zookeeper:zookeeper ches/kafka

这就是我现在运行 zookeeper 和 kafka 容器的方式。指定端口号后，它不再起作用。因为容器实际上是一个孤立的进程。这意味着容器认为它拥有所有的硬件。但实际上并非如此。将端口号指定为 9092 不会将 9092 端口提供给容器。

在后台操作系统将 9092 与物理的合适端口匹配。 docker ps 可以看到。

在上图中，您可以看到0.0.0.0:32769->9092/tcp 表示容器实际上使用了32769 端口。因此，在代码中将端口号更改为 32769 后，它运行良好。希望它可以帮助某人。