kafka 初探

引言

kafka早有耳闻，未尝一用，目前实际使用过rabbitmq和阿里云工作队列datahup，这次在实际项目中终于使用上了。根据项目要求，将订单流水对象进行mysql和es的双写，由es对外提供查询服务，数据库作为数据备份。
我这边当即想出两个策略：

通过kafka，把订单数据推送进入kafka，然后消费者分组监听，实现mysql和es的双写
通过插件，让插件监听mysql的bin日志，由插件进行数据同步(公司有个项目目
前使用的就是这种方式)

最终还是选择了方式一，在对比多方插件的配置情况下和项目存在kafka的情况下，再使用插件进行数据同步显得有些多余(有部分插件同步方式提供消息队列进行异步同步)

引入问题

kafka具有异步、消峰、解耦的效果，但是引入看样子是很不错，但是实际使用起来也需要考虑其他问题

消息重复消费，如何处理

数据双写，先说数据库消息对象中存在订单号，是由雪花算法生成的，同时把订单号设置为mysql主键id，再根据shardingjdbc配置的分库、分表策略，我们把一天的数据统一放在一张表里，通过数据库来实现数据库的消费者的重复消费；
es 通过RestHighLevelClient向es写入数据，存在相同数据他更新，不存在相同数据他就更新，所以说重复消费在当前项目很容易解决

消息消费不掉阻塞消费，又该如何处理

kafka中的消费者是按照索引按顺序消费，这个索引未消费确认以前，是不会进行下一个索引的消费，这个倒是和datahup的消费方式类似。在默认配置情况下，消费首次消费失败后，他会继续再尝试9次，如果接连10次都消费失败，他就会舍弃当前消息，从而进行下一个索引的消费，他这么设计感觉是没问题，但是，他舍弃了消息，就代表这个消息无人处理了，就会造成项目中mysql和es的数据不统一，这是个大麻烦呀！好在他提供了修改默认配置的方法，我的策略是一个消息重复消费5次，每一次消费失败后延迟消费一段时间，如果5次都消费失败，就将其推入死信队列中，通知运营人员去处理消费失败的消息，人工介入。

下面直接看部分代码吧

引入依赖

<!--引入kafka依赖-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

添加配置文件

  kafka:
    bootstrap-servers: 192.168.10.237:9092,192.168.10.238:9092,192.168.10.239:9092
    template:
      default-topic: kfpt-dev
    listener:
      ack-model: MANUAL_IMMEDIATE
    death-topic: ${spring.kafka.template.default-topic}.DLX

生产者代码


import com.alibaba.fastjson.JSON;
import com.xtm.platform.sharding.generator.entity.TOrderDetail;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Component
@Slf4j
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.template.default-topic}")
    private String topicUser;

    /**
     * 发送用户消息
     *
     * @param tOrderDetail 用户信息
     */
    public void sendMessage(TOrderDetail tOrderDetail) {
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topicUser, JSON.toJSONString(tOrderDetail));
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

            @Override
            public void onSuccess(SendResult<String, String> result) {
                RecordMetadata metadata = result.getRecordMetadata();
                log.info("message sent to " + metadata.topic() + ", partition " + metadata.partition() + ", offset " + metadata.offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.info(("send message failed with " + ex.getMessage()));
            }

        });
    }
}

消费者代码

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xtm.platform.sharding.generator.entity.TOrderDetail;
import com.xtm.platform.sharding.service.OrderDetailService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @version : 1.0
 * @description: java类作用描述
 * @author: tianwen
 * @create: 2021/5/11 17:18
 **/
@Component
@Slf4j
public class KafkaSqlConsumer {

    @Autowired
    private OrderDetailService orderDetailService;

    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;

    @Value("${spring.kafka.death-topic}")
    private String deathTopic;

    @Value("${spring.kafka.template.default-topic}")
    private String topic;

    private final String groups = "cs";

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Bean("sqlKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> containerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        //设置提交偏移量的方式
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        SeekToCurrentErrorHandler seekToCurrentErrorHandler = new SeekToCurrentErrorHandler((consumerRecord, e) -> {
            log.error("异常.抛弃这个消息============,{}", consumerRecord.toString(), e);
            kafkaTemplate.send(deathTopic, JSON.toJSONString(DeathMessage.builder().topic(topic).message(consumerRecord.value().toString()).desc("同步sql").createAt(new Date()).build()));
        }, new FixedBackOff(15000L, 5L));
        factory.setErrorHandler(seekToCurrentErrorHandler);
        return factory;
    }


    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>(10);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groups);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //设置每次接收Message的数量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }


    @KafkaListener(topics = {"${spring.kafka.template.default-topic}"}, containerFactory = "sqlKafkaListenerContainerFactory", groupId = groups, concurrency = "3")
    public void consumerMsg(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            log.info("自动topic是: {}, offset是: {}, value是: {}", record.topic(), record.offset(), record.value());
            TOrderDetail tOrderDetail = JSONObject.parseObject(record.value(), TOrderDetail.class);
            orderDetailService.addOrderDetail(tOrderDetail);
            ack.acknowledge();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("kafka push to es fail,topic: {} group : {} body: {} ", record.topic(), "cs", record.value());
        }
    }
}

死信队列消费者


import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * @version : 1.0
 * @description: 死信队列
 * @author: tianwen
 * @create: 2021/5/11 17:18
 **/
@Component
@Slf4j
public class KafkaDeathConsumer {

    @Value("${spring.kafka.death-topic}")
    private String deathTopic;

    @Value("${spring.kafka.template.default-topic}")
    private String topic;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;

    private final String groups = "death";

    @Bean("deathKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> containerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        //设置提交偏移量的方式
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        SeekToCurrentErrorHandler seekToCurrentErrorHandler = new SeekToCurrentErrorHandler((consumerRecord, e) -> {
            log.error("推入死信队列.抛弃这个消息============,{}", consumerRecord.toString(), e);
            kafkaTemplate.send(deathTopic, JSON.toJSONString(DeathMessage.builder().topic(topic).message(consumerRecord.value().toString()).desc("死信队列").createAt(new Date()).build()));
        }, new FixedBackOff(15000L, 5L));
        factory.setErrorHandler(seekToCurrentErrorHandler);
        return factory;
    }


    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>(10);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groups);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //设置每次接收Message的数量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @KafkaListener(topics = {"${spring.kafka.death-topic}"}, containerFactory = "deathKafkaListenerContainerFactory", groupId = groups, concurrency = "1")
    public void consumerMsg(ConsumerRecord record, Acknowledgment ack) {
        log.info("自动topic是: {}, offset是: {}, value是: {}", record.topic(), record.offset(), record.value());
        DingUtil.pushMsgToDing("");
        ack.acknowledge();
    }
}

最后编辑于：2021.06.16 09:39:47

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 214,837评论 6赞 496
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 91,551评论 3赞 389
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 160,417评论 0赞 350
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 57,448评论 1赞 288
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,524评论 6赞 386
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,554评论 1赞 293
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,569评论 3赞 414
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 38,316评论 0赞 270
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,766评论 1赞 307
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 37,077评论 2赞 330
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 39,240评论 1赞 343
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,912评论 5赞 338
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,560评论 3赞 322
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 31,176评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,425评论 1赞 268
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 47,114评论 2赞 366
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 44,114评论 2赞 352