import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class producer {
public static void main(String[] args) {
Properties properties = new Properties();
Properties props = new Properties();
props.put("bootstrap.servers", "hadoop01:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 50; i++) {
producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), "hello world-" + i));
}
producer.close();
}
}
- 打开消费者
kafka-console-consumer.sh --zookeeper hadoop01:2181 --topic first
- 运行api代码,结果如下
Using the ConsoleConsumer with old consumer is deprecated and will be removed
in a future major release. Consider using the new consumer by passing
[bootstrap- server] instead of [zookeeper].
hello world-0
hello world-1
hello world-2
hello world-3
hello world-4
hello world-5
hello world-6
hello world-7
hello world-8
hello world-9
hello world-10
hello world-11
hello world-12
hello world-13
hello world-14
hello world-15
hello world-16
hello world-17
hello world-18
hello world-19
hello world-20
hello world-21
hello world-22
hello world-23
hello world-24
hello world-25
hello world-26
hello world-27
hello world-28
hello world-29
hello world-30
hello world-31
hello world-32
hello world-33
hello world-34
hello world-35
hello world-36
hello world-37
hello world-38
hello world-39
hello world-40
hello world-41
hello world-42
hello world-43
hello world-44
hello world-45
hello world-46
hello world-47
hello world-48
hello world-49