手拉手springboot整合kafka发送消息

@Test
void kafkaSendProducerRecordTest1() {
//参数 String topic, Integer partition, Long timestamp, K key, V value, Iterable
 headers
Headers headers = new RecordHeaders();
headers.add("msg","123".getBytes(StandardCharsets.UTF_8));
ProducerRecord record = new ProducerRecord(
"kafkaTopic01",
0,
System.currentTimeMillis(),
"key",
"hello kafka send message");
kafkaTemplate.send(record);
}

默认主题发送消息

yml配置默认主题

template:
default-topic: default-topic

@Test
void kafkaSendDefaultTest01(){
kafkaTemplate.sendDefault(0,System.currentTimeMillis(),"key01","hello ");
}

发送Object消息

序列化默认为String

@Resource
private KafkaTemplate kafkaTemplate1;
@Test
void kafkaSendObject(){
MessageM messageM =MessageM.builder().userID(123).sn("xo1111").desc("测试").build();
//分区是null，kafka自行决定消息发送到哪个分区
kafkaTemplate1.sendDefault(null,System.currentTimeMillis(),"key01",messageM);
}

Replica副本

Replica副本：为实现备份公共，保证集群中的某个节点发生故障时，确保节点上的partition数据不丢失，且kafka仍然能够正常运行。
Replica副本分为leader Replica和Follower Replica
leader：每个分区多个副本中的主副本，生产者发送数据以及消费者消费数据都来说leader副本。
Follower：每个分区多个副本中的从副本，实时从leader副本中同步数据，保持和leader副本数据同步，当leader副本发送故障，节点中的某个Follower副本会变成新的leader副本。

指定topic分区及副本

• 通过脚本命令创建topic时指定分区和副本

--replication-factor需要小于等于节点个数，不能为0，默认为1

--replication-factor 1表示只有本身

--replication-factor 2 表示本身+副本

./kafka-topics.sh --create --topic testTopic --partitions 3 --replication-factor 2 --bootstrap-server 127.0.0.1:9092

• 代码指定分区及副本

配置bean

@Configuration
public class kafkaConfig {
 
    @Bean
    public NewTopic newTopic(){
        return new NewTopic("topic1", 3, (short) 2);
    }
}

分区策略

Kafak根据不同策略将数据分配到不同的分区

• 默认分配策略：BuiltlnPartitioner
• 轮询分配策略：RoundRobinPartitioner 接口：Partitioner

@Configuration
public class kafkaConfig {
 
    //读取application.yml   bootstrap-servers
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
 
    //读取application.yml   bootstrap-servers
    @Value("${spring.kafka.producer.value-serializer}")
    private String valueSerializer;
 
    @Value("${spring.kafka.producer.key-serializer}")
    private String keySerializer;
 
    @Bean
    public KafkaTemplate kafkaTemplate(){
        return new KafkaTemplate<>(producerFactory());
    }
 
    public Map producerConfigs(){
        Map props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,valueSerializer);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,keySerializer);
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class);
        return props;
    }
 
    /**
     * 生产者工厂
     * @return
     */
    public ProducerFactory producerFactory(){
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
 
}

• 手动指定

• 自定义策略：自定义类实现Partitioner接口；

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。