消息系统之 Kafka (02)

JAVA 生产信息

导入依赖

<!-- 导入 0.10.2 版本 Kafka -->
<dependency>
    <groupId>org.apache.Kafka</groupId>
    <artifactId>Kafka-clients</artifactId>
    <version>0.10.2.0</version>
</dependency>Copy to clipboardErrorCopied

配置生产者

在创建 Producer 对象前，必须配置以下属性：

开发者还可以选择配置如下属性：

通过读取配置，即可生成 Producer 对象。

Properties KafkaProps = new Properties();
KafkaProps.put("bootstrap.servers", "broker1:port1, broker2:port2");
KafkaProps.put("key.serializer", "org.apache.Kafka.common.StringSerializer");
KafkaProps.put("value.serializer", "org.apache.Kafka.common.StringSerializer");
producer = new KafkaProducer<String, String>(KafkaProps);Copy to clipboardErrorCopied

构造消息

实例化 ProducerRecord 类得到消息对象。

创建时必须指定消息所属 Topic 和消息值 Value 。消息发往哪个 Partition 通常由负载均衡机制随机选择。若指定了 Partition 则发送到指定的 Partition，如果没有指定 Partition 但指定了 Key，则由 hasy(key) 决定。

由于 Kafka 只能保证 Partition 内消息的有序性，如果需要保证消息有序到达，Producer 必须指定消息到达的 Partition ，这些消息最终只能被 ConsumeGroup 内的一个 Consumer 消费。

// 三种构造方法
ProducerRecord<String, String> record = new ProducerRecord<>(topic, value);
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
ProducerRecord<String, String> record = new ProducerRecord<>(topic, partition, key, value);

// 发送给消息系统
producer.send(record);Copy to clipboardErrorCopied

接收 ACK

发送消息后，生产者有两种方式接收消息系统返回的 ACK :

1. 通过返回的 Future 判断已经发送成功，get 方法会阻塞线程。实现同步等待。

try {
    Future future = producer.send(record); 
    future.get(10000);
} catch (TimeoutException e) {
    e.printStackTrace();
}Copy to clipboardErrorCopied

1. 发送消息时传递一个回调对象，实现 Kafka 的 Callback 接口，通过回调判断是否发送成功。实现异步等待。

producer.send(record, new ProducerCallback());

private class ProducerCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null)  e.printStackTrace();
    }
} Copy to clipboardErrorCopied

生产示例

import java.util.Properties;
import org.apache.Kafka.clients.producer.KafkaProducer;
import org.apache.Kafka.clients.producer.ProducerConfig;
import org.apache.Kafka.clients.producer.ProducerRecord;
import org.apache.Kafka.common.serialization.StringSerializer;

public class Producer {
    public static String topic = "test"; 

    public static void main(String[] args) throws InterruptedException {

        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.23.76:9092,192.168.23.77:9092");          
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);       
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);    
        p.put("request.required.acks", "-1");                        
        p.put("producer.type", "async");         

        KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(p);

        try {
            for(int i = 0; i < 100; i++) {
                String msg = "Hello," + i;
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);   
                KafkaProducer.send(record);                                  
                Thread.sleep(500);
            }
        } finally {
            KafkaProducer.close();
        }

    }
}Copy to clipboardErrorCopied