手拉手springboot整合kafka发送消息

创建topic时，若不指定topic的分区(Partition主题分区数)数量使，则默认为1个分区(partition)

springboot加入依赖kafka

<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>

加入spring-kafka依赖后，springboot自动装配好kafkaTemplate的Bean

application.yml配置连接kafka

spring:
kafka:
bootstrap-servers: 192.168.68.133:9092

生产者

发送消息

@Resource
private KafkaTemplate<String,String> kafkaTemplate;

@Test
void kafkaSendTest(){
kafkaTemplate.send("kafkamsg01","hello kafka");
}

消费者

接收消息

@Component
public class KafkaConsumer {

@KafkaListener(topics = {"kafkamsg01","test"},groupId = "123")
public void consume(String message){
System.out.println("接收到消息："+message);
}

}

若没有配置groupid

Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is java.lang.IllegalStateException: No group.id found in consumer config, container properties, or @KafkaListener annotation; a group.id is required when group management is used.

@Component
public class KafkaConsumer {

@KafkaListener(topics = {"kafkamsg01","test"},groupId = "123")
public void consume(String message){
System.out.println("接收到消息："+message);
}

}

想从第一条消息开始读取(若同组的消费者已经消费过该主题，并且kafka已经保存了该消费者组的偏移量，则设置auto.offset.reset设置为earliest不生效，需要手动修改偏移量或使用新的消费者组)

application.yml需要将auto.offset.reset设置为earliest

spring:
kafka:
bootstrap-servers: 192.168.68.133:9092
consumer:
auto-offset-reset: earliest

Earliest:将偏移量重置为最早的偏移量

Latest: 将偏移量重置为最新的偏移量

None: 没有为消费者组找到以前的偏移量，向消费者抛出异常

Exception: 向消费者抛出异常

重置消费者组偏移量

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group 123 --topic kafkamsg01 --reset-offsets --to-earliest –execute

重置完成

Spring-kafka生产者发送消息

.send与sendDefault()方法都返回CompletableFuture<String<k,v>>;

CompletableFuture类用于异步编程，表示异步计算结果。该特征使得调用者不必等待操作完成就可以继续执行其他任务，从而提高引用的响应速度和吞吐量

@Resource
private KafkaTemplate<String,String> kafkaTemplate;

@Test
void kafkaSendTest(){
kafkaTemplate.send("kafkamsg01","hello kafka");
}

发送Message

@Test
void kafkaSendMessageTest1(){
//通过构建器模式创建Message
Message<String> message = MessageBuilder.withPayload("hello kafka send message")
.setHeader(KafkaHeaders.TOPIC,"kafkamsg01")
.build();
kafkaTemplate.send(message);
}

SendProducerRecord

String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers

@Test
void kafkaSendProducerRecordTest1() {
//参数 String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers
Headers headers = new RecordHeaders();
headers.add("msg","123".getBytes(StandardCharsets.UTF_8));
ProducerRecord<String,String> 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<String,Object> 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<String,?> kafkaTemplate(){
        return new KafkaTemplate<>(producerFactory());
    }

    public Map<String,Object> producerConfigs(){
        Map<String,Object> 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<String,?> producerFactory(){
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

}

• 手动指定

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