import com.sakura.producer.KafkaProducerDemo;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)

@SpringBootTest

public class ProducerTest {

    @Autowired

    private KafkaProducerDemo kafkaProducerDemo;

    @Test

    public void send() throws InterruptedException {

        String topic = "firstTopic";

        for (int i = 0; i < 6; i++) {

            kafkaProducerDemo.sendMsg(topic, "Hello kafka," + i);

        }

        Thread.sleep(Integer.MAX_VALUE);

    }

}

server.port=8082

#kafka节点

spring.kafka.bootstrap-servers=192.168.204.139:9092

#consumer消息签收机制

spring.kafka.consumer.enable-auto-commit=false

spring.kafka.listener.ack-mode=manual

#如果没有设置offset或者设置的offset不存在时（例如数据被删除）采取的策略：

#earliest：使用最早的offset

#latest：使用最新的offset

#none：使用前一个offset，如果没有就向consumer抛异常

#anything else：直接向consumer抛出异常

spring.kafka.consumer.auto-offset-reset=earliest

## 序列化配置

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

#监听消息消费的线程数，值范围在[1,partitionCounts]之间.

#假如有3个partition,concurrency的值为4,@KafkaListener的数量为2.

#其中一个@KafkaListener会启动两个线程分配到两个partition

#另一个@KafkaListener会启动一个线程分配到另一个partition

#当有一个@KafkaListener挂掉之后会触发broker的再均衡，由剩余的@KafkaListener启动线程重新分配至partition.

#@KafkaListener就像是消费者一样的存在，当值为1时broker会认为只有一个消费者在消费topic.

spring.kafka.listener.concurrency=1

import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.consumer.Consumer;

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

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.kafka.support.Acknowledgment;

import org.springframework.stereotype.Component;

@Slf4j

@Component

public class KafkaConsumerDemo {

    @KafkaListener(topics = "firstTopic",groupId = "groupDemo")

    public void receiveMsg(ConsumerRecord<String, Object> record,

                           Acknowledgment acknowledgment, Consumer<?, ?> consumer) {

        log.info("消费消息：{}", record.value());

        //手动ack

        acknowledgment.acknowledge();

        consumer.commitAsync();

    }

}