学习卡夫卡0.8.2

Question

有一点时间找到一些简单的入门样本与新版本.像KafkaProducer这样的东西与Producer示例不同,并且Web上的大部分旧代码似乎都没有编译相同.

任何指导？Apache Kafka站点没有Java中的生成器示例.

请指教.

Answer 1

在下面的示例中,我使用String作为键创建生成器,使用byte []作为消息内容.

使用基本参数创建新的生产者:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "server1:123,server2:456");
props.put(ProducerConfig.RETRIES_CONFIG, "3");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 200);
props.put(ProducerConfig.BLOCK_ON_BUFFER_FULL_CONFIG, true);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");

KafkaProducer<String, byte[]> producer = new KafkaProducer<String, byte[]>(props);

同步发送消息:

producer.send(new ProducerRecord<>(topic, msgKey, msgContent)).get();

异步发送消息:

producer.send(new ProducerRecord<>(topic, msgKey, msgContent));

您的maven依赖关系对消费者和生产者有利.如果您只需要生产者,您可以使用:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.0</version>
</dependency>

请注意,新的Consumer API可用但暂时不可用.在代码源中,新API将返回null或抛出异常.

Answer 2

试着看看下面的例子:https: //github.com/CameronGregory/kafka/blob/master/TestProducer.java

Answer 3

由于缺乏好的例子，我不得不退步。

这是我的 pom.xml 的一部分

    <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.10</artifactId>
            <version>0.8.2.0</version>
            <scope>compile</scope>
    </dependency>

这是我的代码 - 等待一些测试。

    // KafkaProducer.java - A first pass to verify that we can bring in the appropriate
    //                   libraries using Maven

    // Supports unit tests
    package com.bruno;

    import org.junit.Test;
    import static org.junit.Assert.*;



    import java.util.Arrays;
    import java.util.Properties;

    import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;



    public class MyKafkaProducer
    {

        public static void main(String[] args) throws Exception {
            Properties props = new Properties();

            props.put("metadata.broker.list", "192.168.1.203:9092");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            props.put("request.required.acks", "1");
            ProducerConfig config = new ProducerConfig(props);

            Producer p = new Producer<String, String>(config);

            //sending...
            String topic = "test";
            String message = "Hello Kafka";
            KeyedMessage<String, String> keyedMessage = new KeyedMessage<String, String>(topic, message);
            p.send(keyedMessage);

        }

    }