springboot集成kafka

kafka介紹

Kafka本身是Scala編寫的，運行在JVM之上。Producer和Consumer都通過Kafka的客戶端使用網絡來與之通信。從邏輯上講，Kafka設計非常簡單，它只有一種類似JMS的Topic的消息通道：

 

 

那么Kafka如何支持十萬甚至百萬的並發呢？答案是分區。Kafka的一個Topic可以有一個至多個Partition，並且可以分布到多台機器上：

 

 

Kafka只保證在一個Partition內部，消息是有序的，但是，存在多個Partition的情況下，Producer發送的3個消息會依次發送到Partition-1、Partition-2和Partition-3，Consumer從3個Partition接收的消息並不一定是Producer發送的順序，因此，多個Partition只能保證接收消息大概率按發送時間有序，並不能保證完全按Producer發送的順序。這一點在使用Kafka作為消息服務器時要特別注意，對發送順序有嚴格要求的Topic只能有一個Partition。

Kafka的另一個特點是消息發送和接收都盡量使用批處理，一次處理幾十甚至上百條消息，比一次一條效率要高很多。

最后要注意的是消息的持久性。Kafka總是將消息寫入Partition對應的文件，消息保存多久取決於服務器的配置，可以按照時間刪除（默認3天），也可以按照文件大小刪除，因此，只要Consumer在離線期內的消息還沒有被刪除，再次上線仍然可以接收到完整的消息流。這一功能實際上是客戶端自己實現的，客戶端會存儲它接收到的最后一個消息的offsetId，再次上線后按上次的offsetId查詢。offsetId是Kafka標識某個Partion的每一條消息的遞增整數，客戶端通常將它存儲在ZooKeeper中。

有了Kafka消息設計的基本概念，我們來看看如何在Spring Boot中使用Kafka。

環境准備

　　centos7，idea，jdk1.8+

　　安裝kafka

　　　　下載：

wget https://mirrors.bfsu.edu.cn/apache/kafka/2.4.1/kafka_2.11-2.4.1.tgz

 

 　　　　解壓：

tar -zxvf kafka_2.11-2.4.1.tgz

 　　　　進入：

 

 　　　　修改server.properties

[root@localhost config]# pwd
/home/lpg/kafka/kafka_2.11-2.4.1/config
vi server.properties

#以下需要關注紅色部分，尤其是地址，最好填寫真實的ip
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://我的ip:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/home/lpg/kafka/kafka_2.11-2.4.1/kafka-logs

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log unless the remaining
# segments drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=zk機器ip:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000


############################# Group Coordinator Settings #############################

# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
# The default value for this is 3 seconds.
# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
group.initial.rebalance.delay.ms=0

　　安裝zk

　　　下載zk：zookeeper-3.4.12.tar.gz

　　　解壓：tar -zxvf apache-zookeeper-3.6.2-bin.tar.gz

　　　進入zk的conf目錄：cd conf/

　　　拷貝生成zoo.cfg：cp zoo_sample.cfg zoo.cfg

　　　修改zoo.cfg文件：

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/home/lpg/zookeeper/apache-zookeeper-3.6.2-bin/data
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#此處可配置集群節點，此處忽略

　#server.1=172.17.3.205:2888:3888

　#server.2=172.17.3.206:2888:3888

　#server.3=172.17.3.207:2888:3888

　　安裝KafkaOffsetMonitor

　　　　KafkaOffsetMonitor是Kafka的一款客戶端消費監控工具，用來實時監控Kafka服務的Consumer以及它們所在的Partition中的Offset，我們可以瀏覽當前的消費者組，並且每個Topic的所有Partition的消費情況都可以一目了然。

　　　　下載jar包：KafkaOffsetMonitor-assembly-0.2.0.jar 即可

啟動各個服務

　　按照順序啟動如下服務：

　　zk：進入bin目錄，執行：./zkServer.sh start

　　kafka：進入bin目錄，執行：./kafka-server-start.sh ../config/server.properties

　　KafkaOffsetMonitor：進入jar包所在目錄，執行：java -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk localhost:2181 --port 8089 --refresh 10.seconds --retain 1.days

 與springboot整合

　　pom依賴

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

　　application.yml配置

server:
  port: 7889

spring:
  kafka:
    bootstrap-servers: 172.22.3.14:9092

    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
      properties:
        linger.ms: 1

    consumer:
      enable-auto-commit: false
      auto-commit-interval: 100ms
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        session.timeout.ms: 15000
      group-id: test-group-id

　　解釋下以上屬性含義：

bootstrap-servers：連接kafka的地址，多個地址用逗號分隔
batch-size：當將多個記錄被發送到同一個分區時， Producer 將嘗試將記錄組合到更少的請求中。這有助於提升客戶端和服務器端的性能。這個配置控制一個批次的默認大小（以字節為單位）。16384是缺省的配置
retries：若設置大於0的值，客戶端會將發送失敗的記錄重新發送
buffer-memory：Producer 用來緩沖等待被發送到服務器的記錄的總字節數，33554432是缺省配置
key-serializer：關鍵字的序列化類
value-serializer：值的序列化類

 

　　生產與消費

package com.lpg.kafka.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author lpg
 * @description: kfk生產者與消費者
 * @date 2020-12-2317:57
 */
@Service
@Slf4j
public class KfkService {
    @Autowired
    private KafkaTemplate<Integer,String> kafkaTemplate;


    //消費者：監聽topic1，groupId1
    @KafkaListener(topics = {"topic1"},groupId = "groupId1")
    public void consumer1(ConsumerRecord<Integer,String> record){
        log.info("consumer1 kfk consume message start...");
        log.info("consumer1 kfk consume message topic:{},msg:{}",record.topic(),record.value());
        log.info("consumer1 kfk consume message end...");
    }
    //消費者：監聽topic1，groupId2
    @KafkaListener(topics = {"topic1"},groupId = "groupId2")
    public void consumer3(ConsumerRecord<Integer,String> record){
        log.info("consumer3 kfk consume message start...");
        log.info("consumer3 kfk consume message topic:{},msg:{}",record.topic(),record.value());
        log.info("consumer3 kfk consume message end...");
    }
    //消費者：監聽topic1，groupId2
    @KafkaListener(topics = {"topic1"},groupId = "groupId2")
    public void consumer2(ConsumerRecord<Integer,String> record){
        log.info("consumer2 kfk consume message start...");
        log.info("consumer2 kfk consume message topic:{},msg:{}",record.topic(),record.value());
        log.info("consumer2 kfk consume message end...");
    }

　　//生產者
    public void sendMsg(String topic , String msg){
        log.info("開始發送kfk消息,topic:{},msg:{}",topic,msg);

        ListenableFuture<SendResult<Integer, String>> sendMsg = kafkaTemplate.send(topic, msg);
        //消息確認
        sendMsg.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("send error,ex:{},topic:{},msg:{}",throwable,topic,msg);
            }

            @Override
            public void onSuccess(SendResult<Integer, String> stringStringSendResult) {
                log.info("send success,topic:{},msg:{}",topic,msg);
            }
        });
        log.info("kfk send end!");
    }

}

　　　　測試

package com.lpg.kafka.controller;

import com.lpg.kafka.service.KfkService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author lpg
 * @description: 測試kfk生產與消費
 * @date 2020-12-2318:22
 */
@RestController
public class KfkController {
    @Autowired
    private KfkService kfkService;
    @GetMapping("/send")
    public String send(){
        kfkService.sendMsg("topic1","I am topic msg");
        return "success-topic1";
    }
}

　　　　啟動idea服務

　　　　　　啟動之后，瀏覽器輸入http://localhost:7889/send

　　　　　　運行結果如下：

 

 

 

 從上面測試結果，可以印證：同一group的topic只允許一個線程來消費。

monitor插件監控

　　瀏覽器輸入http://ip:8089/#/