0


探究:kafka生产者/消费者与多线程安全

1. 多线程安全

1.1. 生产者是多线程安全的么?

    Kafka生产者是线程安全的,可以在多个线程中共享一个Kafka生产者实例。这是因为Kafka生产者实例内部使用了一些同步机制来保证线程安全,例如使用了线程安全的队列来缓存消息,使用了同步锁来保护共享资源的访问等。

    同时,Kafka生产者的send()方法是非阻塞的,可以在多个线程中并发调用,不会阻塞线程。Kafka生产者还提供了异步发送和同步发送两种发送方式,可以根据实际需求选择不同的发送方式。

然而,如果多个线程共享同一个Kafka生产者实例,需要注意以下几点:

  1. 同一个线程中不要同时调用send()方法和flush()方法,可能会导致消息发送顺序不一致。

  2. 不同线程中调用send()方法时,需要注意消息的顺序,可以使用Kafka的分区机制来保证消息的顺序。

  3. 如果多个线程发送的消息都是针对同一个主题或分区,可能会导致消息的重复发送或丢失。因此,建议在多线程情况下,使用Kafka的分区机制,将消息发送到不同的分区中,以避免消息的重复和丢失。

     综上所述,Kafka生产者是线程安全的,可以在多个线程中共享一个Kafka生产者实例,但需要注意消息的顺序和分区的使用,以保证消息的可靠性和顺序性。
    

1.1. 消费者是多线程安全的么?

    Kafka消费者是非线程安全的,主要原因是因为:Kafka消费者使用了内部的状态来跟踪消费进度和偏移量。这种状态包括消费者的位置,消费者的偏移量,以及消费者的订阅主题和分区等信息。如果多个线程同时访问同一个Kafka消费者实例,就会导致这些状态信息的不一致,从而导致消费进度出现错误。

2. 消费者规避多线程安全方案

背景:Kafka消费者中具有poll()方法,该方法是一个阻塞方法,如果在同一个线程中多次调用poll()方法,会导致消费者阻塞在poll()方法中,无法及时消费新到达的消息。因此,为了提高Kafka消费者的并发性能,通常使用多个消费者线程来消费+处理消息。为了保证Kafka消费者的线程安全性,通常采用以下2种方案。

方案的本质:一个线程只能绑定一个消费者实例(不能多个线程共用一个消费者实例)

2.1. 每个线程维护一个kafkaConsumer

  1. 创建多个线程,去消费topic

  2. 每个线程绑定固定数量的分区(最好的情况是一个消费者绑定一个分区)

    代码比较简单,唯一需要关注的点:创建的消费者线程和分区的数量可能不相等,此时,可以采用「分区副本自动分配策略」,该策略会将消费者线程和分区绑定在一起。

场景讨论:

    按照现在的代码,假设消费者线程获取了分区A 100个msg,然后处理,若还未处理完,此时,offset未提交。因为此时offset还未提交,消费者线程还会去分区A调用poll,获取到的msg依然会是刚才的100个msg。存在消费重复的场景,因此,消费者需要做好幂等处理。

    上面的代码,**不会**出现这个场景:消费者获取了消息1,2,3,3处理完成了,执行了提交偏移,但是消息1还未提交的场景。

代码示例:

package com.bie.kafka.kafkaThrea;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

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

/**
 *
 *       1、KafkaConsumer是非线程安全的,KafkaProducer是线程安全的。
 *       2、该案例是创建多个线程,每个线程维护一个KafkaConsumer实例
 *           用户创建多个线程消费topic数据,使用分区副本自动分配策略,将消费者线程与分区进行绑定
 *       3、ConsumerRunnable,消费线程类,执行真正的消费任务
 */
public class ConsumerRunnable implements Runnable {

    // 每个线程维护私有的kafkaConsumer实例
    private final KafkaConsumer<String, String> consumer;

    /**
     * 默认每个消费者的配置参数初始化
     *
     * @param brokerList
     * @param groupId
     * @param topic
     */
    public ConsumerRunnable(String brokerList, String groupId, String topic) {
        // 带参数的构造方法
        Properties props = new Properties();
        // kafka的列表
        props.put("bootstrap.servers", brokerList);
        // 消费者组编号
        props.put("group.id", groupId);
        // 自动提交
        props.put("enable.auto.commit", true);
        // 提交提交每个一秒钟
        props.put("auto.commit.interval.ms", "1000");
        // 反序列化key
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 反序列化value
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 将配置信息进行初始化操作
        this.consumer = new KafkaConsumer<>(props);
        // 定义响应的主题信息topic:使用分区副本自动分配策略
        consumer.subscribe(Arrays.asList(topic));
    }

    @Override
    public void run() {
        // 消费者保持一直消费的状态
        while (true) {
            // 将获取到消费的信息
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(200));
            // 遍历出每个消费的消息
            for (ConsumerRecord<String, String> record : records) {
                // 处理消息:deal msg
            }
            // 提交offset
        }
    }
}
package com.bie.kafka.kafkaThrea;

import java.util.ArrayList;
import java.util.List;

/**
 *       1、消费线程管理类,创建多个线程类执行消费任务
 */
public class ConsumerGroup {

    // 消费者群组,多消费者。
    private List<ConsumerRunnable> consumers;

    public ConsumerGroup(int consumerNum, String groupId, String topic, String brokerList) {
        // 初始化消费者组
        consumers = new ArrayList<>(consumerNum);
        // 初始化消费者,创建多少个消费者
        for (int i = 0; i < consumerNum; i++) {
            // 根据消费者构造方法,创建消费者实例
            ConsumerRunnable consumerRunnable = new ConsumerRunnable(brokerList, groupId, topic);
            // 将创建的消费者实例添加到消费者组中
            consumers.add(consumerRunnable);
        }
    }

    public void execute() {
        // 将消费者组里面的消费者遍历出来
        for (ConsumerRunnable task : consumers) {
            // 创建一个消费者线程,并且启动该线程
            new Thread(task).start();
        }
    }
}
package com.bie.kafka.kafkaThrea;

public class ConsumerMain {

    public static void main(String[] args) {
        // kafka即broker列表
        String brokerList = "slaver1:9092,slaver2:9092,slaver3:9092";
        // group组名称
        String groupId = "group1";
        // topic主题名称
        String topic = "topic1";
        // 消费者的数量
        int consumerNum = 3;
        // 通过构造器创建出一个对象
        ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
        // 执行execute的方法,创建出ConsumerRunnable消费者实例。多线程多消费者实例
        consumerGroup.execute();
    }
}

上面代码,在实际工程应用中存在问题,描述见下:

  1. 每个消费者线程,先执行poll拉取一批batchsize命令后

  2. 批量处理这批消息

  3. 提交offset

     步骤2,处理batchsize的消息,可能中间的某一个失败了,但是步骤3提交了整体的offset,会导致失败的消息丢失了。
    

解决方案:处理每个消息的流程中,增加重试机制(本地消息表),保证该消息能执行成功

2.2. [单/多]kafkaConsumer实例 + 多worker线程

与2.1方案一的区别在于,将「消息的获取」与「消息的处理」解耦开

  1. 消息的获取:维护一个or多个kafkaConsumer实例,获取消息,获取到消息后,将消息丢到消息处理线程池中

  2. 消息的处理:创建一个线程池,里面存放了worker线程,每个worker线程处理获取到的消息

ConsumerWorker类

    worker线程:执行msg的处理,更新offsets信息
package huxi.test.consumer.multithreaded;
 
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
 
import java.util.List;
import java.util.Map;
 
public class ConsumerWorker<K, V> implements Runnable {
 
    private final ConsumerRecords<K, V> records;
    private final Map<TopicPartition, OffsetAndMetadata> offsets;
 
    public ConsumerWorker(ConsumerRecords<K, V> record, Map<TopicPartition, OffsetAndMetadata> offsets) {
        this.records = record;
        this.offsets = offsets;
    }
 
    @Override
    public void run() {
        for (TopicPartition partition : records.partitions()) {
            // 获取到分区的消息记录
            List<ConsumerRecord<K, V>> partitionRecords = records.records(partition);
            // 遍历获取到的消息记录
            for (ConsumerRecord<K, V> record : partitionRecords) {
                // 消息处理逻辑
            }
 
            /* 下面操作,是更新各个分区的offset信息到offsets变量,并没有真正的提交位移 */
            /* 真正的更新位移操作,不是在worker线程,而是在消费者线程 */

            // 待更新的位移
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
            // 同步锁,锁住offsets位移
            synchronized (offsets) {
                // 如果offsets位移不包含partition这个key信息,就将位移信息设置到map集合里面
                if (!offsets.containsKey(partition)) {
                    offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
                } else {
                    // 否则,offsets位移包含partition这个key信息,获取到offsets的位置信息
                    long curr = offsets.get(partition).offset();
                    if (curr <= lastOffset + 1) { // 如果获取到的位置信息小于等于上一次位移信息大小,将这个partition的位置信息设置到map集合中。并保存到broker中。
                        offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
                    }
                }
            }
        }
    }
}

ConsumerThreadHandler类

    一个主线程,定时去poll消息,然后将消息投递到worker线程中,最后,当offsets信息发生变更后,提交offset
package huxi.test.consumer.multithreaded;
 
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
 
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
public class ConsumerThreadHandler<K, V> {
 
    // KafkaConsumer实例
    private final KafkaConsumer<K, V> consumer;
    // ExecutorService实例
    private ExecutorService executors;
    // 位移信息offsets
    private final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
 
    public ConsumerThreadHandler(String brokerList, String groupId, String topic) {
        // 构造kafkaConsumer配置
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "false"); // 非自动提交位移信息
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        // 创建kafkaConsumer,赋值给成员变量consumer
        consumer = new KafkaConsumer<>(props);

        // 消费者订阅消息,并实现重平衡rebalance
        // rebalance监听器,创建一个匿名内部类。使用rebalance监听器前提是使用消费者组(consumer group)。
        // 监听器最常见用法就是手动提交位移到第三方存储以及在rebalance前后执行一些必要的审计操作。
        consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
            // 在coordinator开启新一轮rebalance前onPartitionsRevoked方法会被调用。
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                consumer.commitSync(offsets); // 提交位移
            }

            // rebalance完成后会调用onPartitionsAssigned方法。
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                offsets.clear(); // 清除位移信息
            }
        });
    }
 
    /**
     * 消费主方法
     * @param threadNumber  线程池中线程数
     */
    public void consume(int threadNumber) {
        // 创建一个worker线程池,线程数量为threadNumber个
        executors = new ThreadPoolExecutor(
                threadNumber,
                threadNumber,
                0L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(1000),
                new ThreadPoolExecutor.CallerRunsPolicy());

        // 只有一个消费者线程
        try {
            // 消费者一直处于等待状态,等待消息消费
            while (true) {
                // 从主题中获取消息
                ConsumerRecords<K, V> records = consumer.poll(1000L);
                // 如果获取到的消息不为空
                if (!records.isEmpty()) {
                    // submit: 将一批msg交给worker线程处理,消息处理完后,更新offsets信息
                    executors.submit(new ConsumerWorker<>(records, offsets));
                }
                // 调用提交位移信息
                commitOffsets();
            }
        } catch (WakeupException e) {
            // swallow this exception
        } finally {
            commitOffsets();  // 调用提交位移信息
            consumer.close(); // 关闭consumer
        }
    }
 
    private void commitOffsets() {
        Map<TopicPartition, OffsetAndMetadata> unmodfiedMap;
        // 保证线程安全、同步锁,锁住offsets
        synchronized (offsets) {
            // 判断如果offsets位移信息为空,直接返回,节省同步锁对offsets的锁定的时间
            if (offsets.isEmpty()) {
                return;
            }
            // 如果offsets位移信息不为空,将位移信息offsets放到集合中,方便同步
            unmodfiedMap = Collections.unmodifiableMap(new HashMap<>(offsets));
            // 清除位移信息offsets
            offsets.clear();
        }
        // 将封装好的位移信息unmodfiedMap集合进行同步提交
        // 手动提交位移信息
        consumer.commitSync(unmodfiedMap);
    }
 
    public void close() {
        consumer.wakeup();
        // 关闭ExecutorService实例
        executors.shutdown();
    }
}

Main类

    包装了所有的工具类,启动整个程序
package huxi.test.consumer.multithreaded;
 
public class Main {
 
    public static void main(String[] args) {
        // broker列表、topic、group id
        String brokerList = "localhost:9092";
        String topic = "test-topic";
        String groupID = "test-group";
        
        // 1. 根据ConsumerThreadHandler构造方法,创建出消费者handler
        final ConsumerThreadHandler<byte[], byte[]> handler = new ConsumerThreadHandler<>(brokerList, groupID, topic);
        final int cpuCount = Runtime.getRuntime().availableProcessors();
 
        // 创建线程的匿名内部类
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                // 执行consume,在此线程中执行消费者消费消息。
                handler.consume(cpuCount); 
            }
        };
        new Thread(runnable).start();  // 3. 该函数会调用上面的run()方法
 
        try {
            // 20秒后自动停止该测试程序
            Thread.sleep(20000L);
        } catch (InterruptedException e) {
            // swallow this exception
        }
        System.out.println("Starting to close the consumer...");
        handler.close();
    }
}  

2.3.方案优缺点对比

优点缺点方案1
实现简单

速度较快,因为无线程交互开销

方便位移管理

易于维护分区之间的消息消费顺序

socket连接开销大

consumer受限于topic的分区数,扩展性差

broker端处理负载高(因为发往broker的请求较多)

reblance可能性增大
方案2
消息获取和处理解耦

可独立扩展consumer和worker数,伸缩性好

实现负载

难于维护分区内的消息顺序

处理链路变长,导致位移管理困难

worker线程异常可能导致消费数据丢失

标签: kafka java 分布式

本文转载自: https://blog.csdn.net/weixin_36750623/article/details/129441996
版权归原作者 粪逗er 所有, 如有侵权,请联系我们删除。

“探究:kafka生产者/消费者与多线程安全”的评论:

还没有评论