生产者和消费者

生产者API

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;


public class ProducerMain {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.4.121.218:9092");
        /**
         * acks 控制着认定请求已经完成的标准,all是最慢却最耐用的设置
         */
        props.put("acks", "all");
        /**
         * 如果请求失败了,生产者可以自动重新发送请求的次数
         */
        props.put("retries", 0);
        /**
         * 生产者为每一个partition还没发送出去的记录设置一个缓存区,此配置设置了缓存区的大小,
         * 调大参数会导致更大的批处理,但是需要更多的内存,因为会为每一个激活的partition创建
         * 这么大的缓存区
         */
        props.put("batch.size", 16384);
        /**
         * 默认情况下,即使缓存区中还有空间,也可以立即发送请求。如果要减少请求次数,可以设置
         * linger.ms > 0,这样生产者再发送请求之前会等待该毫秒数,希望有更多的记录被添加到缓
         * 存区以填满这个批次的缓存。但是如果1ms之后,仍然没有记录被添加进来,每个请求就会白
         * 白增加1ms的延迟。所以在高负载下,可以用一小部分延迟来换取更少、但更高效的请求。另
         * 外,即便linger.ms = 0,在一段时间内到达的记录也是会被同一批次处 理的
         */
        props.put("linger.ms", 1);
        /**
         * 生产者可以使用的缓存空间的总量。当请求的发送速度比请求发送到服务器的速度还快,那么缓
         * 存空间很快就会被耗尽,这时其他的请求就会被阻塞。阻塞的时长由max.block.ms决定,如果
         * 阻塞的时间超过max.block.ms的值,会抛出TimeoutException
         */
        props.put("buffer.memory", 33554432);
        /**
         * serializer指定了如何将ProducerRecord的键值对对象转换成字节。可以使用
         * ByteArraySerializer 或者 StringSerializer
         */
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        /**
         * 生产者包含一个缓冲池,它保存还未传送到集群上的记录,以及一个后台I/O线程,负责将记录转
         * 换成请求,并将其传输到集群。使用后没能正常关闭生产者会泄露缓冲池中的资源。
         */
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 0; i < 100; i++) {
            /**
             *  send()方法是异步的,调用时会将记录添加到待处理记录的缓冲池中,然后立即返回
             *  这使得生产者可以对缓冲池中的各个记录进行批处理,从而提高效率
             */
            producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i)));
        }
        producer.close();
    }
    
}

消费者API

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

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

public class AutomaticOffsetConsumer {

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

        /**
         * 通过bootstrap.servers指定要连接的brokers地址,这个列表仅用于发现集群中的其他代理,
         * 并不一定是集群中所有服务器的详尽地址列表。尽管你可能想要指定多个以防客户端连接的时候
         * 存在服务器故障。
         */
        props.put("bootstrap.servers", "10.4.121.218:9092");
        /**
         *   kafka用consumer group的概念,允许一组进程对消费和处理记录的工作进行划分。这些
         * 进程要么运行在同一台机器上,要么运行在高可扩展和容错的分布式集群上。所有group.id相同
         * 的消费者实例属于同一个consumer group。
         * 
         *   同组中的consumer可以通过 subscribe API 来动态订阅话题列表。然后,kafka会把被订阅
         * 话题中的每一条消息分发给每个consumer group的一个进程。这是通过平衡同组所有成员的分区来
         * 实现的,这样每个分区正好能分到组中的一个消费者。如果一个主题有4个分区,一个consumer group
         * 有2个进程,那么一个进程就消费两个分区。
         * 
         *   consumer group中的成员资格是动态维护的:如果进程失败了,原本分配给它的partitions
         * 将重新分给同组中的其他进程。类似的,如果一个新的consumer被添加到consumer group中,
         * partitions将会被重新分配。这叫group rebalance。当新的partition被添加到被订阅的topic
         * 中或新建与订阅的正则表达式匹配的topic时,也会发生group rebalance。group通过定期的元数
         * 据刷新来检测是否有新的partitions,并且将新的partitions分配给组中的成员。
         * 
         *   在概念上,可以将consumer group看作是由多个进程组成的逻辑订阅者。作为一个多订阅用户
         * 系统,就可以支持任意多个consumer group,对于不同的consumer group,可以重复消费消息,
         * 只要订阅主题就可以了。
         */
        props.put("group.id", "test");
        
        // 自动提交已消费的`offset`  
        props.put("enable.auto.commit", "true");
        // 通过`auto.commit.interval.ms`来控制提交`offset`的频率
        props.put("auto.commit.interval.ms", "1000");
        /**
         *  当kafka中没有初始`offset`或如果当前`offset`在服务器上不再存在时,共支持三种消费策略。
         *  注意不论那种策略,如果有当前`group.id`在`topic`的`partition`下有提交的`offset`
         *  则从提交的`offset`开始消费
         *  
         *  earliest: 当无提交的`offset`时,从分区的最开始消费
         *  latest(default): 当无提交的`offset`时,从分区下最新生产的的数据开始消费 
         *  none: 只要有一个分区不存在已提交的`offset`,则抛出异常
         */
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        // 订阅两个主题 foo bar
        consumer.subscribe(Arrays.asList("foo", "bar"));
        System.out.println("主题订阅:foo bar");
        while (true) {
            /**
             *   订阅完主题之后,当调用poll方法时,consumer会被自动添加到group中。poll()是用来确保消费者
             * 存活的方法,只要你一直调用poll方法,consumer就会一直存在group中,接收分配给这个 consumer 的
             * partitions中的消息。consumer 会每隔一段时间就发送一次心跳,如果这个 consumer 崩溃了或者超过
             * session.timeout.ms设置的时间没有过发送心跳,那么这个consumer就被认为挂了,分配给这个consumer
             * 的 partitions 也会被重新分配。
             * 
             *   也有可能consumer一直发送心跳,但是不调用poll去处理消费信息,这样consumer会永远占用分配给它的
             * partitions,为了防止这种情况,kafka提供了一种存活检测机制, max.poll.interval.ms。即如果你
             * 调用poll()的时间间隔超过了这个值,客户端就会主动将consumer移除出group,然后重新分配partitions。
             * 这个机制确保了只有活跃的consumer才能够提交offset,所以为了consumer能保留在group中,你需要一直
             * 调用poll方法。
             * 
             *   有两种方法控制poll轮询
             * 
             * 1、max.poll.interval.ms:提高调用poll的最大间隔时间,这样就可以给consumer更多时间去处理从
             * poll中获取的这一批记录。这样做的缺点是会延迟组的rebalance,因为consumer只有在调用poll的时候
             * 才被添加到group,然后rebalance。你可以通过这个参数来控制rebalance的频率,但是当实际调用poll
             * 的频率达不到这个参数的要求,就会被移出group,这样就会使得处理速度会很慢。
             * 
             * 2、max.poll.records:调用一次poll最多返回的记录数。这样就能更容易的去预测一次poll间隔时间内
             * 要处理多少条记录。调小这个参数值,就可以减少间隔时间,从而减少rebalance的影响。
             * 
             *   对于批次消息处理时间不确定的情况,这些选项都不够。推荐方法是将消息处理移动到另一个线程(异步),
             * 这样消费者就能在处理器处理记录的同时调用poll方法。但是这样就需要禁用自动提交,在线程处理完数据
             * 后手动提交记录的offset。需要注意的是提交的offset不能领先实际的position。另外还需要暂停分区,
             * 以便在线程处理完之前的记录之后再接受从poll返回的新纪录。
             */
            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());
        }
    }
}

打赏一个呗

取消

感谢您的支持,我会继续努力的!

扫码支持
扫码支持
扫码打赏,一毛也是爱

打开支付宝扫一扫,即可进行扫码打赏哦