kafka_2.12-1.1.0 生产与消费java实现示例

环境准备：

1）需要在maven工程中引入依赖：

 <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.12</artifactId>
      <version>1.1.0</version>
    </dependency>
    <dependency>
      <groupId> org.apache.cassandra</groupId>
      <artifactId>cassandra-all</artifactId>
      <version>0.8.1</version>

      <exclusions>
        <exclusion>
          <groupId>org.slf4j</groupId>
          <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
          <groupId>log4j</groupId>
          <artifactId>log4j</artifactId>
        </exclusion>
      </exclusions>

    </dependency>

2）本机是否能telnet 192.178.0.111 9092（kafaka所部署的vmw虚拟机）通？ 如果telnet端口不通，则需要关闭192.178.0.111的防火墙：

systemctl stop firewalld.service #停止firewall
systemctl disable firewalld.service #禁止firewall开机启动

一、生产者

首先看以下两种实现示例：

package com.dx;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.Date;

public class ProducerTest {
    public static void main(String[] args) {
        producer_test1(args);

        producer_test2();
    }

    private static void producer_test2() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.178.0.111: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<String, String>(props);
        for(int i = 0; i < 10; i++)
            producer.send(new ProducerRecord<String, String>("kafakatopic", Integer.toString(i), Integer.toString(i)));

        producer.close();
    }

    private static void producer_test1(String[] args) {
        String arg0 = args != null && args.length > 0 ? args[0] : "10";
        long events = Long.parseLong(arg0);
        Random rnd = new Random();

        //    /opt/kafka_2.12-1.1.0/bin/kafka-console-producer.sh --broker-list 192.178.0.111:9092 --sync --topic kafkatopic
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.178.0.111: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");
        // 配置partitionner选择策略，可选配置
        props.put("partitioner.class", "com.dx.SimplePartitioner2");

        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (long nEvents = 0; nEvents < events; nEvents++) {
            long runtime = new Date().getTime();
            String ip = "192.178.0." + rnd.nextInt(255);
            String msg = runtime + ",www.example.com," + ip;
            ProducerRecord<String, String> data = new ProducerRecord<String, String>("kafakatopic", ip, msg);
            Future<RecordMetadata> send = producer.send(data,
                    new Callback() {
                        public void onCompletion(RecordMetadata metadata, Exception e) {
                            if (e != null) {
                                e.printStackTrace();
                            } else {
                                System.out.println("The offset of the record we just sent is: " + metadata.offset());
                            }
                        }
                    });
        }
        producer.close();
    }
}

SimplePartitioner2.java

package com.dx;

import java.util.List;
import java.util.Map;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

public class SimplePartitioner2 implements Partitioner {
    public void configure(Map<String, ?> map) {
    }

    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        int partition = 0;
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        String stringKey = (String) key;
        int offset = stringKey.lastIndexOf('.');
        if (offset > 0) {
            partition = Integer.parseInt(stringKey.substring(offset + 1)) % numPartitions;
        }

        return partition;
    }

    public void close() {
    }
}

参数设置备注：

1）bootstrap.servers --设置生产者需要连接的kafka地址
2）acks --回令类型
3）retries --重试次数
4）batch.size --批量提交大小
5）linger.ms --提交延迟等待时间（等待时间内可以追加提交）
6）buffer.memory --缓存大小
7）key.serializer|value.serializer --序列化方法

需要注意的有两点:
1、acks回令。如果必须等待回令，那么设置acks为all；否则，设置为-1；等待回令会有性能损耗。
2、生产者在发送消息的过程中，会自己默认批量提交。所以，如果单条指令的发送请求，记得发送完后flush才能生效。

3、SimplePartitioner2.java为kafaka分区，可选项。

二、消费者

以下实现示例：

package com.dx;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;
import java.util.Arrays;
import java.util.Map;
import java.util.HashMap;
import java.util.List;

/**
 * zk启动：sh /opt/zookeeper-3.4.11/bin/zkServer.sh start &
 * kafka启动：sh /opt/kafka_2.12-1.1.0/bin/kafka-server-start.sh /opt/kafka_2.12-1.1.0/config/server.properties &
 */
public class ConsumerTest {
    public static void main(String[] args) {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.178.0.111:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG ,"test") ;
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("kafakatopic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

三、测试

先启动productor运行，之后在启动consumer运行。在consumer打印结果如下：

offset = 200, key = 192.178.0.20, value = 1522587459181,www.example.com,192.178.0.20
offset = 201, key = 192.178.0.143, value = 1522587459359,www.example.com,192.178.0.143
offset = 202, key = 192.178.0.113, value = 1522587459359,www.example.com,192.178.0.113
offset = 203, key = 192.178.0.110, value = 1522587459359,www.example.com,192.178.0.110
offset = 204, key = 192.178.0.232, value = 1522587459359,www.example.com,192.178.0.232
offset = 205, key = 192.178.0.96, value = 1522587459359,www.example.com,192.178.0.96
offset = 206, key = 192.178.0.76, value = 1522587459360,www.example.com,192.178.0.76
offset = 207, key = 192.178.0.78, value = 1522587459360,www.example.com,192.178.0.78
offset = 208, key = 192.178.0.80, value = 1522587459360,www.example.com,192.178.0.80
offset = 209, key = 192.178.0.177, value = 1522587459360,www.example.com,192.178.0.177
offset = 210, key = 0, value = 0
offset = 211, key = 1, value = 1
offset = 212, key = 2, value = 2
offset = 213, key = 3, value = 3
offset = 214, key = 4, value = 4
offset = 215, key = 5, value = 5
offset = 216, key = 6, value = 6
offset = 217, key = 7, value = 7
offset = 218, key = 8, value = 8
offset = 219, key = 9, value = 9