Apache Kafka 简单生产者示例

由 bjcl 创建，youj 最后一次修改 2016-12-27 让我们使用Java客户端创建一个用于发布和使用消息的应用程序。 Kafka生产者客户端包括以下API。KafkaProducer API让我们了解本节中最重要的一组Kafka生产者API。 KafkaProducer API的中心部分是 KafkaProducer 类。 KafkaProducer类提供了一个选项，用于将其构造函数中的Kafka代理连接到以下方法。KafkaProducer类提供send方法以异步方式将消息发送到主题。 send()的签名如下producer.send(new ProducerRecord<byte[],byte[]>(topic, partition, key1, value1) , callback); ProducerRecord - 生产者管理等待发送的记录的缓冲区。回调 - 当服务器确认记录时执行的用户提供的回调(null表示无回调)。KafkaProducer类提供了一个flush方法，以确保所有先前发送的消息都已实际完成。 flush方法的语法如下 - public void flush()KafkaProducer类提供了partitionFor方法，这有助于获取给定主题的分区元数据。 这可以用于自定义分区。 这种方法的签名如下 - public Map metrics()它返回由生产者维护的内部度量的映射。public void close() - KafkaProducer类提供关闭方法块，直到所有先前发送的请求完成。生产者API生产者API的中心部分是生产者类。 生产者类提供了一个选项，通过以下方法在其构造函数中连接Kafka代理。生产者类生产者类提供send方法以使用以下签名向单个或多个主题发送消息。public void send(KeyedMessaget<k,v> message) - sends the data to a single topic,par-titioned by key using either sync or async producer.public void send(List<KeyedMessage<k,v>>messages)- sends data to multiple topics.Properties prop = new Properties();prop.put(producer.type,"async")ProducerConfig config = new ProducerConfig(prop);有两种类型的生产者 - 同步和异步。相同的API配置也适用于同步生产者。 它们之间的区别是同步生成器直接发送消息，但在后台发送消息。 当您想要更高的吞吐量时，异步生产者是首选。 在以前的版本，如0.8，一个异步生产者没有回调send()注册错误处理程序。 这仅在当前版本0.9中可用。public void close()生产者类提供关闭方法以关闭与所有Kafka代理的生产者池连接。配置设置下表列出了Producer API的主要配置设置，以便更好地理解 - S.No配置设置和说明1 client.id 标识生产者应用程序2 producer.type 同步或异步3 acks acks配置控制生产者请求下的标准是完全的。4重试如果生产者请求失败，则使用特定值自动重试。5bootstrapping代理列表。6 linger.ms 如果你想减少请求的数量，你可以将linger.ms设置为大于某个值的东西。7 key.serializer 序列化器接口的键。8 value.serializer 值。9 batch.size 缓冲区大小。10 buffer.memory 控制生产者可用于缓冲的存储器的总量。ProducerRecord APIProducerRecord是发送到Kafka cluster.ProducerRecord类构造函数的键/值对，用于使用以下签名创建具有分区，键和值对的记录。public ProducerRecord (string topic, int partition, k key, v value)主题 - 将附加到记录的用户定义的主题名称。分区 - 分区计数。键 - 将包含在记录中的键。值 - 记录内容。public ProducerRecord (string topic, k key, v value)ProducerRecord类构造函数用于创建带有键，值对和无分区的记录。主题 - 创建主题以分配记录。键 - 记录的键。值 - 记录内容。public ProducerRecord (string topic, v value)ProducerRecord类创建一个没有分区和键的记录。主题 - 创建主题。值 - 记录内容。ProducerRecord类方法列在下表中 - S.No类方法和描述1 public string topic()主题将附加到记录。2 public K key()将包括在记录中的键。 如果没有这样的键，null将在这里重新打开。3 public V value()记录内容。4 partition()记录的分区计数SimpleProducer应用程序在创建应用程序之前，首先启动ZooKeeper和Kafka代理，然后使用create topic命令在Kafka代理中创建自己的主题。 之后，创建一个名为 Sim-pleProducer.java 的java类，然后键入以下代码。//import util.properties packagesimport java.util.Properties;//import simple producer packagesimport org.apache.kafka.clients.producer.Producer;//import KafkaProducer packagesimport org.apache.kafka.clients.producer.KafkaProducer;//import ProducerRecord packagesimport org.apache.kafka.clients.producer.ProducerRecord;//Create java class named “SimpleProducer"public class SimpleProducer { public static void main(String[] args) throws Exception{ // Check arguments length value if(args.length == 0){ System.out.println("Enter topic name"); return; } //Assign topicName to string variable String topicName = args[0].toString(); // create instance for properties to access producer configs Properties props = new Properties(); //Assign localhost id props.put("bootstrap.servers", “localhost:9092"); //Set acknowledgements for producer requests. props.put("acks", “all"); //If the request fails, the producer can automatically retry, props.put("retries", 0); //Specify buffer size in config props.put("batch.size", 16384); //Reduce the no of requests less than 0 props.put("linger.ms", 1); //The buffer.memory controls the total amount of memory available to the producer for buffering. props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serializa-tion.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serializa-tion.StringSerializer"); Producer<String, String> producer = new KafkaProducer <String, St