我已经在 Windows 机器中的 Eclipse 中编写了 kafka 生产者并进行了分区。我的 Kafka 集群在 ec2 linux 中运行。我能够从 Eclipse 执行 kafaka 生产者代码，但我没有看到 ec2 框中的主题。
Produce code :
package com.panda.kafka.training;
import java.util.*;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class PandaKafkaProducer {
public static void main(String[] args) {
long events = Long.parseLong(args[0]);
Random rnd = new Random();
Properties props = new Properties();
props.put("metadata.broker.list", "ec2-xx-yy-zzz-212.compute-1.amazonaws.com:9092");
//props.put("producer.type", "sync");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("partitioner.class", "com.panda.kafka.training.PandaKafkaPartitioner");
props.put("request.required.acks", "1");
props.put("producer.type","async");
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<String,
String>(config);
for (long nEvents = 0; nEvents < events; nEvents++)
{ System.out.println("creating event "+nEvents);
long runtime = new Date().getTime();
String ip = "192.168.2."+ rnd.nextInt(255);
String msg = runtime + ",www.vulab.com," + ip; KeyedMessage<String, String> data = new KeyedMessage<String, String>("vulab123", ip, msg);
producer.send(data);
}
producer.close();
}
}
Server Properties file:
# The port the socket server listens on
port=9092
# Hostname the broker will bind to. If not set, the server will bind to all interfaces
host.name=localhost
# Hostname the broker will advertise to producers and consumers. If not set, it uses the
# value for "host.name" if configured. Otherwise, it will use the value returned from
# java.net.InetAddress.getCanonicalHostName().
advertised.host.name=ec2-xx-yy-zzz-212.compute-1.amazonaws.com
# The port to publish to ZooKeeper for clients to use. If this is not set,
# it will publish the same port that the broker binds to.
advertised.port=0000
Producer properties file:
# list of brokers used for bootstrapping knowledge about the rest of the cluster
# format: host1:port1,host2:port2 ...
metadata.broker.list=52.2.202.212:0000
# name of the partitioner class for partitioning events; default partition spreads data randomly
#partitioner.class=
# specifies whether the messages are sent asynchronously (async) or synchronously (sync)
producer.type=sync
Output from Producer in ec2:
[ec2-user@ip-xxxx bin]$ sh kafka-console-producer.sh --broker-list xxxx:yyy.zzzz:9092 --topic vulab123
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
任何详细的解释都会有很大帮助。