分片的存储实体是副本 3 个副本，就是这个分片的数据会存 3 份。

Topic: my-13-topic Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1

Partition: 0 分片号

Leader: 0 leader 副本在 0 号 Broker 上

Replicas: 0,2,1 三个副本位于 0 2 1 号 broker 上

Isr: 0,2,1 in sync 处于同步状态的 broker

分片和 broker 的关系

broker: 一般是一台计算机上一个 Broker

分片： 分片的一个副本存放在一个 broker 上。

分片数 是在创建 主题时 根据你的估算数据量来选择的。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic

监控管理工具?

Kakfa 自身未提供图形化的监控管理工具，市面上有很多开源的监控管理工具，但都不怎么成熟可靠。

Kafka Offffset Monitor

kafkaoffsetmonitor · GitHub Topics

• Kafka集群状态
• Topic、Consumer Group列表
• bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning
• --topic my-replicated-topic图形化展示topic和consumer之间的关系
• 图形化展示consumer的Offffset、Lag等信息

java -cp KafkaOffsetMonitor-assembly-0.2.1.jar \ com.quantifind.kafka.offsetapp.OffsetGetterWeb \ 
--offsetStorage kafka --zk zk-server1,zk-server2 \ 
--port 8080 \ 
--refresh 10.seconds \ 
--retain 2.days

java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \ com.quantifind.kafka.offsetapp.OffsetGetterWeb \ 
--zk zk-server1,zk-server2 \
--port 8088 \ 
--refresh 10.seconds \ 
--retain 2.days

启动后就可以在浏览器中访问了： http://localhost:8088

spring中使用

Spring for Apache Kafka

SpringforApacheKafka（SpringKafka）项目将核心Spring概念应用于基于Kafka的消息传递解决方案的开发。它提供了一个“模板”，作为发送消息的高级抽象。它还支持带有@KafkaListener注释和“侦听器容器”的消息驱动POJO。

这些库促进了依赖项注入和声明性的使用。在所有这些情况下，您将看到与Spring框架中的JMS支持和Spring AMQP中的RabbitMQ支持的相似之处。

• KafkaTemplate

• KafkaMessageListenerContainer

• @KafkaListener

• KafkaTransactionManager

• spring-kafka-test

?在maven项目中引用

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

添加配置，包括序列化器

在spring中配置 kafka配置类。包括提前配置好的 一些事务等等。

?数据消费 创建主题等。

?配置一个消费消费消息 重试三次还不成功，将消息发送到死信队列。

public class Sender {

	public static void main(String[] args) {
		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
		context.getBean(Sender.class).send("test", 42);
	}

	private final KafkaTemplate<Integer, String> template;

	public Sender(KafkaTemplate<Integer, String> template) {
		this.template = template;
	}

	public void send(String toSend, int key) {
		this.template.send("topic1", key, toSend);
	}

}

public class Listener {

    @KafkaListener(id = "listen1", topics = "topic1")
    public void listen1(String in) {
        System.out.println(in);
    }

}

@Configuration
@EnableKafka
public class Config {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String>
                        kafkaListenerContainerFactory(ConsumerFactory<Integer, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProps());
    }

    private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // ...
        return props;
    }

    @Bean
    public Sender sender(KafkaTemplate<Integer, String> template) {
        return new Sender(template);
    }

    @Bean
    public Listener listener() {
        return new Listener();
    }

    @Bean
    public ProducerFactory<Integer, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(senderProps());
    }

    private Map<String, Object> senderProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        //...
        return props;
    }

    @Bean
    public KafkaTemplate<Integer, String> kafkaTemplate(ProducerFactory<Integer, String> producerFactory) {
        return new KafkaTemplate<Integer, String>(producerFactory);
    }

}

监控数据都是 使用KafkaListener 注解? ?监听收到消息

@KafkaListener(id = "multiGroup", topics = { "foos", "bars" })

在运行的时候

?通过spring的思想，都是配置得到。

MessageConverter的调用

?