kafka生产者与消费者

package kafka;


import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.log4j.Logger;

import java.util.Properties;


public class Producer {

    Logger logger = Logger.getLogger("Producer");

    public KafkaProducer getKafkaProducer() {
        Properties kafkaProps = new Properties();
        /**
         * kafka生产者必选是三个属性
         * bootstrap.servers 指定broker的地址清单
         * key.serializer 必须是一个实现org.apache.kafka.common.serialization.Serializer接口的类，将key序列化成字节数组。注意：key.serializer必须被设置，即使消息中没有指定key
         * value.serializer  将value序列化成字节数组
         */

        kafkaProps.put("bootstrap.servers", "localhost:9092");
        kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //其他设置
        /*
        acks=0：如果设置为0，生产者不会等待kafka的响应，高吞吐。消息会被立刻加到发送缓冲通道中，并且认为已经发送成功。这种情况下，不能保证kafka接收到了这条消息，retries配置不会生效，每条消息的偏移量都是1；

        acks=1：这个配置意味着kafka会把这条消息写到本地日志文件中，但是不会等待集群中其他机器的成功响应。这种情况下，在写入日志成功后，集群主机器挂掉，同时从机器还没来得及写的话，消息就会丢失掉。

        acks=all：这个配置意味着leader会等待所有的follower同步完成。这个确保消息不会丢失，除非kafka集群中所有机器挂掉。这是最强的可用性保证，最安全模式，但延迟相对较长。

        （1）acks指定必须要有多少个partition副本收到消息，生产者才会认为消息的写入是成功的。

              acks=0，生产者不需要等待服务器的响应，以网络能支持的最大速度发送消息，吞吐量高，但是如果broker没有收到消息，生产者是不知道的

              acks=1，leader partition收到消息，生产者就会收到一个来自服务器的成功响应

              acks=all，所有的partition都收到消息，生产者才会收到一个服务器的成功响应

        （2）buffer.memory，设置生产者内缓存区域的大小，生产者用它缓冲要发送到服务器的消息。

        （3）compression.type，默认情况下，消息发送时不会被压缩，该参数可以设置成snappy、gzip或lz4对发送给broker的消息进行压缩

        （4）retries，生产者从服务器收到临时性错误时，生产者重发消息的次数

        （5）batch.size，发送到同一个partition的消息会被先存储在batch中，该参数指定一个batch可以使用的内存大小，单位是byte。不一定需要等到batch被填满才能发送

        （6）linger.ms，生产者在发送消息前等待linger.ms，从而等待更多的消息加入到batch中。如果batch被填满或者linger.ms达到上限，就把batch中的消息发送出去

        （7）max.in.flight.requests.per.connection，生产者在收到服务器响应之前可以发送的消息个数
         */
        kafkaProps.put("acks", "all");//
        return new KafkaProducer(kafkaProps);
    }

    /**
     * 同步发送
     *
     * @param topic
     * @param key
     * @param value
     * @param kafkaProducer
     */
    public void sendMsgSynchr(String topic, String key, String value, KafkaProducer kafkaProducer) {

        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, key, value);
        kafkaProducer.send(producerRecord);
    }

    /**
     * 异步发送
     *
     * @param topic
     * @param key
     * @param value
     * @param kafkaProducer
     */
    public void sendMsgAsynchr(String topic, String key, String value, KafkaProducer kafkaProducer) {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic, key, value);
        kafkaProducer.send(producerRecord, new ProducerCallback());//发送消息时，传递一个回调对象，该回调对象必须实现org.apahce.kafka.clients.producer.Callback接口
    }

    private class ProducerCallback implements Callback {

        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {//如果Kafka返回一个错误，onCompletion方法抛出一个non null异常。
                e.printStackTrace();//对异常进行一些处理，这里只是简单打印出来
            }
        }
    }

    public static void main(String[] args) {
        Producer producer = new Producer();
        KafkaProducer kafkaProducer = producer.getKafkaProducer();
        for (int i = 0; i < 100; i++) {
            String msg = "msg------" + i;
            System.out.println(msg);
            producer.sendMsgAsynchr("test_kafka", null, msg, kafkaProducer);
        }
    }
}

package kafka;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Collections;
import java.util.Properties;

public class Consumer {


    public KafkaConsumer getKafkaConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "groupid1");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //其他参数
        /*
        1：fetch.min.bytes，指定消费者从broker获取消息的最小字节数，即等到有足够的数据时才把它返回给消费者

        2：fetch.max.wait.ms，等待broker返回数据的最大时间，默认是500ms。fetch.min.bytes和fetch.max.wait.ms哪个条件先得到满足，就按照哪种方式返回数据

        3：max.partition.fetch.bytes，指定broker从每个partition中返回给消费者的最大字节数，默认1MB

        4：session.timeout.ms，指定消费者被认定死亡之前可以与服务器断开连接的时间，默认是3s

        5：auto.offset.reset，消费者在读取一个没有偏移量或者偏移量无效的情况下（因为消费者长时间失效，包含偏移量的记录已经过时并被删除）该作何处理。默认是latest（消费者从最新的记录开始读取数据）。另一个值是             　      earliest（消费者从起始位置读取partition的记录）

        6：enable.auto.commit，指定消费者是否自动提交偏移量，默认为true

        7：partition.assignment.strategy，指定partition如何分配给消费者，默认是Range。Range：把Topic的若干个连续的partition分配给消费者。RoundRobin：把Topic的所有partition逐个分配给消费者

        8：max.poll.records，单次调用poll方法能够返回的消息数量
         */
        return new KafkaConsumer(props);
    }

    public void getMsg(String topic, KafkaConsumer kafkaConsumer) {
        //2.订阅Topic

        //创建一个只包含单个元素的列表，Topic的名字叫作customerCountries
        kafkaConsumer.subscribe(Collections.singletonList(topic));  //主题列表
        //支持正则表达式，订阅所有与test相关的Topic
        //consumer.subscribe("test.*");
        //3.轮询
        //消息轮询是消费者的核心API，通过一个简单的轮询向服务器请求数据，一旦消费者订阅了Topic，轮询就会处理所欲的细节，包括群组协调、partition再均衡、发送心跳
        //以及获取数据，开发者只要处理从partition返回的数据即可。

//        try {
        while (true) {//消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环
            //在100ms内等待Kafka的broker返回数据.超市参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回

            ConsumerRecords<String, String> records = kafkaConsumer.poll(100l);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.value());
            }
        }
//        } finally {
//            //退出应用程序前使用close方法关闭消费者，网络连接和socket也会随之关闭，并立即触发一次再均衡
//            kafkaConsumer.close();
//        }
    }

    public static void main(String[] args) {
        Consumer consumer = new Consumer();
        KafkaConsumer kafkaConsumer = consumer.getKafkaConsumer();
        consumer.getMsg("test_kafka", kafkaConsumer);
    }
}