开心一刻

今天小学女同学给我发消息
她：你现在是毕业了吗
我：嗯，今年刚毕业
她给我发了一张照片，怀里抱着一只大橘猫
她：我的眯眯长这么大了，好看吗
我：你把猫挪开点，它挡住了，我看不到
她：你是 sb 吗，滚
我解释道：你说的是猫呀
可消息刚发出，就出现了红色感叹号，并提示：消息已发出，但被对方拒收了

你也没说猫叫眯眯呀

kafka搭建

出于简单考虑，基于
docker
搭建一个
kafka
节点；因为一些原因，国内的 Docker Hub 镜像加速器都不可用了，目前比较靠谱的做法是搭建个人镜像仓库，可参考：
Docker无法拉取镜像解决办法
，我已经试过了，是可行的，但还是想补充几点

  1. sync-image-example.yml
    只需要修改最后的镜像拷贝，其他内容不需要改


    sync-image-example_改动点

    支持一次配置多个镜像的拷贝

  2. 镜像拷贝

    docker 镜像拷贝命令的格式


    skopeo copy docker://docker.io/命名空间/镜像名:TAG docker://阿里云镜像地址/命名空间/镜像名:TAG


    我们以 kafka 为例，去 Docker Hub 一搜，好家伙，搜出来上万个


    上万个kafka镜像

    我们将搜索条件精确化一些，搜
    wurstmeister/kafka


    wurstmeister_kafka

    点进去，它在 Docker Hub 的地址是：


    https://hub.docker.com/r/wurstmeister/kafka


    那它的 docker 地址就是


    docker://docker.io/wurstmeister/kafka


    其他的镜像用类似的方式去找，所以最终的拷贝命令类似如下：


    skopeo copy docker://docker.io/wurstmeister/kafka:latest docker://registry.cn-hangzhou.aliyuncs.com/qingshilu/wurstmeister_kafka:latest


    如果一切顺利，那么在我们的阿里云个人镜像仓库就能看到我们拷贝的镜像了


    阿里云个人镜像

  3. 如何 pull

    在个人仓库点镜像名，会看到
    操作指南


    如何pull

    我们只关注前两步，就可以将镜像 pull 下来


    镜像过滤_wurstmeister

镜像获取到之后，就可以搭建
kafka
了；因为依赖
zookeeper
，我们先启动它

docker run -d --name zookeeper-test -p 2181:2181 \
--env ZOO_MY_ID=1 \
-v zookeeper_vol:/data \
-v zookeeper_vol:/datalog \
-v zookeeper_vol:/logs \
registry.cn-hangzhou.aliyuncs.com/qingshilu/wurstmeister_zookeeper

然后启动
kafka 

docker run -d --name kafka-test -p 9092:9092 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.2.118:2181 \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.2.118 \
--env KAFKA_ADVERTISED_PORT=9092  \
--env KAFKA_LOG_DIRS=/kafka/logs \
-v kafka_vol:/kafka  \
registry.cn-hangzhou.aliyuncs.com/qingshilu/wurstmeister_kafka

不出意外的话，都启动成功

kafka启动成功

如果出意外了，大家也别慌，用
docker log
去查看日志，然后找对应的解决方案

# 1.先找到启动失败的容器id
docker ps -a
# 2.用 docker log 查看容器启动日志
docker log 容器id

如果需要开启
kafka
的
SASL
认证，可参考：
Docker-Compose搭建带SASL用户密码验证的Kafka
来搭建

Kafka Tool

详情可查看：
kafka可视化客户端工具（Kafka Tool）的基本使用

kafka_tool连接成功

创建 Topic：
test-topic
，并发送一条消息

创建test_topic并发送一条消息

此时
test-topic
中有 1 条消息

消费者 poll

代码很简单

/**
 * @author: 青石路
 */
public class MsgConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(MsgConsumer.class);

    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.2.118:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test_group");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 如果在kafka中找不到当前消费者的偏移量，则设置为最旧的
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
        // 订阅主题
        consumer.subscribe(Collections.singleton("test-topic"));
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        LOGGER.info("records count = {}", records.count());
        records.forEach(record -> LOGGER.info("{} - {} - {}", record.offset(), record.key(), record.value()));
        // consumer.commitAsync();
        consumer.close();
    }
}

我们执行下，输出日志如下

poll没拉取到消息_日志

竟然 poll 不到消息，为什么呀？

思考考

我们调整下代码，循环 poll

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    LOGGER.info("records count = {}", records.count());
    records.forEach(record -> LOGGER.info("{} - {} - {}", record.offset(), record.key(), record.value()));
}

我们再执行下，输出日志如下

while循环后日志

消费者 poll 的过程中会先判断当前消费者是否在
消费者组
中，如果不在，会先加入消费者组，在加入过程中，
ConsumerCoordinator
会对这个消费者组 Rebalance，整个过程中该消费者组内的所有消费者都不能工作，而 poll 又配置了超时时间（
100 毫秒
），如果在超时时间内，当前消费者还未正常加入消费者组中，那么 poll 肯定是拉取不到数据的；根据日志可以看出，第 3 次 poll 的时候，消费者已经正常加入消费者组中，那么就能 poll 到数据了

很多小伙伴可能可能会有这样的疑问

平时在项目中使用的时候，从来没有感受到这样的问题，为什么呢

原因有以下几点

  1. poll 的超时时间设置比较长，超时时间内消费者能够正常加入到消费者组中
  2. 消费者随项目的启动创建，存活周期与项目一致，那么只有前几次 poll 的时候，可能会因为消费者未加入到消费者组中而拉取不到数据，而一旦消费者成功加入到消费者组之后，那么只要 Topic 中有数据，poll 肯定能拉取到数据；从整个次数占比来看，poll 拉取不到数据的异常情况（Topic 中有可拉取的数据，但 poll 不到）占比非常小，小到可以忽略不计了

所以你们感受不到这样；但如果某些场景下，比如 DataX 从 kafka 读数据

异源数据同步 → DataX 为什么要支持 kafka？

消费者要不断新建，那么 poll 不到数据的异常情况的占比就会上来了，那就需要通过一些机制来降低其所造成的的影响了，比如说重试机制

总结

  1. 示例代码：
    kafka-demo
  2. 如果大家平时用 docker 比较多，推荐通过搭建个人镜像仓库来解决镜像拉取超时的问题
  3. kakfa 消费者 poll 的时候，消费者如果不在消费者组中，会先加入消费者组，那么超时时间内可能 poll 不到数据，可以通过增大超时时间，或者重试机制来降低 poll 不到数据的异常次数（Topic 中没有可拉取的数据而 poll 不到的情况不算异常情况）

标签: none

添加新评论