// 生产40条数据
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class KafkaProducerTest {
// 只有主题分区超过一个以上的，才会由不同的消费者进行消费，一个分区的数据只能由组内的一个消
费者进行消费，本示例的topic有2个分区
public final static String TOPIC_NAME = "t_consumerGroup";
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "server1:9092");
props.put("acks", "all");
props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<String,
String>(props);
boolean flag = true;
int count = 0;
while (flag) {
count++;
for (int i = 0; i < 10; ++i) {
try {
// 获取返回值Future，该对象封装了返回值
Future<RecordMetadata> future = producer.send(new
ProducerRecord<String, String>(TOPIC_NAME, i + "", i + ""));
// 调用一个Future.get()方法等待响应
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
