【消息队列】一文搞定大数据消息队列Kafka3

c-1：topic-0、topic-2、topic-4、topic-6
c-2：topic-1、topic-3、topic-5

c-1：topic-0、topic-1、topic-2、topic-3
c-2：topic-4、topic-5、topic-6

#yum配置文件修改
logging:
  config: classpath:logback.log

<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <!-- 格式化输出： %d表示日期， %thread表示线程名， %-5level: 级别从左显示5个字符宽度 %msg:日志消息, %n是换行符 -->
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS}[%thread] %-5level %logger{50} - %msg%n</pattern>
        </encoder>
    </appender>
    <!--log日志的级别-->
    <root level="info">
        <appender-ref ref="STDOUT" />
    </root>
</configuration>

#消费者组id，分组内的消费者只能消费该消息一次，不同的组可以重复此消息
group.id
#为true的时候自动提交偏移量
enable.auto.commit
#自动提交offset周期
auto.commit.interval.ms
#重置消费便宜量策略，消费者在读取一个没有偏移量的分区或者偏移量无效的情况下（消费者长时间读取一个无效的偏移量记录）该怎么处理
#默认是latest，如果需要从头消费，则需要消费者组名变更才可以
auto.offset.reset
#序列化器，反序列化机制，生产者序列化，消费者反序列化
key.deserializer

public static Properties getProperties() {
        Properties props = new Properties();
        //kafka服务器broker地址
        props.put("bootstrap.servers", "8.140.116.67:9092");
    //消费者组id，同组消费者只能一个消费者消费，不同组可以重复消费消息
        props.put("group.id", "xdclass-g1");
    //默认是latest，从最新的消息开始消费，earliest是从头开始消费，但是还要改消费者组
        props.put("auto.offset.reset","earliest");
    //是否自动提交偏移量，一般不自动
        props.put("enable.auto.commit", "false");
    //自动提交offset延迟时间
        props.put("auto.commit.interval.ms", "1000");
    //反序列化机制
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }

订阅主题api：subscribe(topic主题集合)
拉取消息api：poll(Duration.ofMillis(500))

@Test
    public void simpleConsumerTest(){
        Properties props = getProperties();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //订阅topic主题
        consumer.subscribe(Arrays.asList(KafkaProducerTest.TOPIC_NAME));
       while (true) {
            //拉取时间控制，阻塞超时时间,当拉取topic里没有消息时，消费者阻塞500毫秒
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
            for (ConsumerRecord<String, String> record : records) {
                System.err.printf("topic = %s, offset = %d, key = %s, value = %s%n",record.topic(), record.offset(), record.key(), record.value());
            }
        }
    }

//默认是latest，如果需要从头消费partition消息，需要改为earliest，且变更消费者组名，才生效
props.put("auto.offset.reset","earliest");

 @Test
    public void simpleConsumerTest(){
        Properties properties = getProperties();
        KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<>(properties);
        //订阅主题
        kafkaConsumer.subscribe(Arrays.asList(KafkaProducerTest.TOPIC_NAME));
        while(true){
            ConsumerRecords<String,String> poll = kafkaConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : poll) {
                System.out.println("topic:"+record.topic()+"--"+"key:"+record.key()+"--"+"offset:"+record.offset()+"--"+"value:"+record.value());
            }
            //同步阻塞提交
            //kafkaConsumer.commitSync();
            if(!poll.isEmpty()){
                //异步提交offset
                kafkaConsumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if(exception == null){
                            System.out.println("手工提交offset成功");
                        }else{
                            System.out.println("手工提交offset失败");
                            exception.printStackTrace();
                        }
                    }
                });
            }
        }
    }

index文件中并没有为每一条message简历索引，采用了稀疏存储的方式
每隔一定字节的数据建立一条索引，避免了索引文件占用过多的空间和资源，从而可以将索引文件保留到内存中
缺点是没有建立索引的数据在查询过程中需要小范围的顺序扫描操作

#默认是1G,segment文件达到1G(index和log文件的总和)，进行分片
log.segment.bytes=1073741824

#分段一
00000000000000000000.index 00000000000000000000.log
#分段二 数字1234指的是当前文件的最小偏移量offset，即上个文件的最后一个消息offset+1
00000000000000001234.index 00000000000000001234.log
#分段三
00000000000000088888.index 00000000000000088888.log

CA：如果不要求P（不允许分区），则C（强一致性）和A（可用性）是可以保证的。但放弃P的同时也就意味着放弃了系统的扩展性，也就是分布式节点受限，每办法部署子节点，违背分布式系统设计的初衷。
CP：如果不要求A（可用性），每个请求都需要在服务器之间保持强一致，而P（分区）会导致同步时间无限延长（也就是等待数据同步完成才可以正常访问），一旦发生网络故障或者消息丢是的情况，就要牺牲用户的体验，等待所有数据全部一致了在让去用户去访问
AP：要高可用并且允许分区，则需要放弃一致性。一旦分区发生，节点之间可能会失去联系，为了高可用，每个节点只能用本地数据提供服务，而这样会导致全局数据的不一致性。

tar -xvf apache-zookeeper-3.7.0-bin.tar.gz
mv apache-zookeeper-3.7.0 zk1
cp -r zk1 zk2
cp -r zk1 zk3

#客户端端口
clientPort=2181
#数据存储路径,以节点为命名区分
dataDir=/tmp/zookeeper/2181
#修改AdminServer端口
admin.serverPort=8888

cd /tmp/zookeeper/2181
echo 1 >myid
cd /tmp/zookeeper/2182
echo 2 >myid
cd /tmp/zookeeper/2183
echo 3 >myid

#各个配置文件zoo.cfg加入集群配置
#server.服务器id=服务器IP地址:服务器直接通信的端口:服务器之间选举投票的端口
server.1=127.0.0.1:2881:3881
server.2=127.0.0.1:2882:3882
server.3=127.0.0.1:2883:3883

#启动zk1节点
cd /usr/local/software/zk1/bin
./zkServer.sh start
#启动zk2节点
cd /usr/local/software/zk2/bin
./zkServer.sh start
#启动zk3节点
cd /usr/local/software/zk3/bin
./zkServer.sh start

#内网中使用，内网部署kafka集群只须用到listeners，内网需要作为区分时，才需要用到advertised.listeners
listeners=PLAINTEXT://172.18.123.229:9092
advertised.listeners=PLAINTEXT://112.74.55.160:9092
#每个节点编号1、2、3
broker.id=1
#端口
port=9092
#配置3个
log.dirs=/tmp/kafka-logs/kafka1
#zk地址
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183

./kafka-server-start.sh -daemon ../config/server.properties & #守护线程
./kafka-server-start.sh ../config/server.properties &

./kafka-topics.sh --create --zookeeper localhost:2181,localhost:2182,localhost:2183 --replication-factor 3 --partitions 6 --topic xdclass-cluster-topic

log.retention.check.interval.ms=30000

#清理超过指定时间的消息，默认是168小时，7天
#还有log.retention.ms,log.metention.minutes,log.retention.hours  优先级高到低
log.retention.hours=168
#超过指定大小后，删除旧的消息，下面是1G的字节数，-1就是没有限制
log.retention.bytes=1073741824
#还有基于日志起始位移（log start offset）

每个日志文件都维护一个最大时间戳字段，每次日志写入新的消息时，都会更新该字段
一个日志段，segment写满了被切分之后，时间戳就不会在发生改变，这时kafka可以根据当前的时间去和各个日志文件的时间戳去进行比较来判断是否过期了

超过阈值的部分不许要大于一个日志字段的大小，这回kafka会把最早的一个日志文件删除
假设热值大小是500MB，当前分区共有四个日志文件，大小分别为500MB，500MB，500MB，10MB
假设阈值设置1500，那么kafka会计算当前文件的总值，500*3+10=1510MB，1510-1500=10 <500,所以kafka不会删除日志文件
假设阈值设置1000，1510-1000 = 510 >500,所以kafka会删除最早的一个日志文件

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

 private static final String TOPIC_NAME = "user.register.topic";
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    /**
     * 发送消息
     * @param phone
     */
    @GetMapping("/api/user/{phone}")
    public void sendMessage1(@PathVariable("phone") String phone) {
        kafkaTemplate.send(TOPIC_NAME, phone).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
        }, failure -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        });
    }

- 配置文件修改增加生产者信息
```bash
spring:
  kafka:
    bootstrap-servers: 192.168.111.31:9092,192.168.111.31:9093,192.168.111.31:9094
    producer:
      # 消息重发的次数。
      retries: 0
      #一个批次可以使用的内存大小
      batch-size: 16384
      # 设置生产者内存缓冲区的大小。
      buffer-memory: 33554432
      # 键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all

 private static final String TOPIC_NAME = "user.register.topic";
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    /**
     * 发送消息
     * @param phone
     */
    @GetMapping("/api/user/{phone}")
    public void sendMessage1(@PathVariable("phone") String phone) {
        kafkaTemplate.send(TOPIC_NAME, phone).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
        }, failure -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        });
    }