Обучение Kafka 0.8.2

Ответ 1

В приведенном ниже примере я создаю продюсера, использующего String as key и 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>

Обратите внимание, что новый API-интерфейс доступен, но пока не используется. В исходном коде новый API возвращает исключение null или throw.

Ответ 2

Попробуйте взглянуть на следующий пример: https://github.com/CameronGregory/kafka/blob/master/TestProducer.java

Ответ 3

Всегда полезно проверить, как оригинальные авторы тестируют свой код, чтобы вы могли понять, чего они пытаются достичь или желаемого использования (если и когда есть предоставленные тесты:)

В этом случае просто проверьте этот код: https://github.com/apache/kafka/blob/0.8.2/examples/src/main/java/kafka/examples/Producer.java

:)

Ответ 4

Мне пришлось регрессировать из-за отсутствия хороших примеров.

Вот часть моего 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);

        }

    }