kafka实战

古城微笑少年丶 2022-02-09 10:41 261阅读 0赞

kafka原理请参考上篇博客：[kafka原理详解][kafka]

## java程序中使用kafka ##

**1  创建Topic**

public static void main(String[] args) {
        //创建topic
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.180.128:9092");
        AdminClient adminClient = AdminClient.create(props);
        ArrayList<NewTopic> topics = new ArrayList<NewTopic>();
        NewTopic newTopic = new NewTopic("topic-test", 1, (short) 1);
        topics.add(newTopic);
        CreateTopicsResult result = adminClient.createTopics(topics);
        try {
            result.all().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

使用AdminClient API可以来控制对kafka服务器进行配置，我们这里使用如下的构造方法来创建了一个名为“topic-test”，分区数为1，复制因子为1的Topic.

>     NewTopic(String name, int numPartitions, short replicationFactor)

**2  Producer生产者发送消息**

public static void main(String[] args){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.180.128:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<String, String>("topic-test", Integer.toString(i), Integer.toString(i)));
    
        producer.close();
    
    }

*  bootstrap.servers: 用于初始化时建立链接到kafka集群，以host:port形式，多个以逗号分隔host1:port1,host2:port2；
 *  acks: 生产者需要server端在接收到消息后，进行反馈确认的尺度，主要用于消息的可靠性传输；acks=0表示生产者不需要来自server的确认；acks=1表示server端将消息保存后即可发送ack，而不必等到其他follower角色的都收到了该消息；acks=all(or acks=-1)意味着server端将等待所有的副本都被接收后才发送确认。
 *  retries: 生产者发送失败后，重试的次数
 *  batch.size: 当多条消息发送到同一个partition时，该值控制生产者批量发送消息的大小，批量发送可以减少生产者到服务端的请求数，有助于提高客户端和服务端的性能。
 *  linger.ms: 默认情况下缓冲区的消息会被立即发送到服务端，即使缓冲区的空间并没有被用完。可以将该值设置为大于0的值，这样发送者将等待一段时间后，再向服务端发送请求，以实现每次请求可以尽可能多的发送批量消息。
 *  batch.size和linger.ms是两种实现让客户端每次请求尽可能多的发送消息的机制，它们可以并存使用，并不冲突。
 *  buffer.memory: 生产者缓冲区的大小，保存的是还未来得及发送到server端的消息，如果生产者的发送速度大于消息被提交到server端的速度，该缓冲区将被耗尽。
 *  key.serializer,value.serializer 说明了使用何种序列化方式将用户提供的key和vaule值序列化成字节。

**KafkaProducer对象实例化方法**,可以使用map形式的键值对或者Properties对象来配置客户端的属性

public KafkaProducer(Map<String,Object> configs);
    public KafkaProducer(Map<String,Object> configs, Serializer<K> keySerializer,Serializer<V> valueSerializer);
    public KafkaProducer(Properties properties);
    public KafkaProducer(Properties properties, Serializer<K> keySerializer,Serializer<V> valueSerializer);

**消息发送方法send()：**

public Future<RecordMetadata> send(ProducerRecord<K,V> record); // Equivalent to send(record, null)
    public Future<RecordMetadata> send(ProducerRecord<K,V> record,Callback callback);

send方法负责将缓冲池中的消息异步的发送到broker的指定topic中。异步发送是指，方法将消息存储到底层待发送的I/O缓存后，将立即返回，这可以实现并行无阻塞的发送更多消息。send方法的返回值是RecordMetadata类型，它含有消息将被投递的partition信息，该条消息的offset，以及时间戳。  
       因为send返回的是Future对象，因此在该对象上调用get()方法将阻塞，直到相关的发送请求完成并返回元数据信息；或者在发送时抛出异常而退出。

**3 Consumer消费者消费消息**

public static void main(String[] args){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.12.65:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        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");
        final KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
        consumer.subscribe(Arrays.asList("topic-test");
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }

这里我们使用Consume API 来创建了一个普通的java消费者程序来监听名为“topic-test”的Topic，每当有生产者向kafka服务器发送消息，我们的消费者就能收到发送的消息。

## **SpringBoot整合kafka** ##

**1   基本配置信息**

与其他spring的项目一样，总是离不开配置，这里我们使用java配置来配置我们的kafka消费者和生产者。

引入pom文件：

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

kafka配置文件（properties）

#============== kafka ===================
    kafka.consumer.zookeeper.connect=10.93.21.21:2181
    kafka.consumer.servers=10.93.21.21:9092
    kafka.consumer.enable.auto.commit=true
    kafka.consumer.session.timeout=6000
    kafka.consumer.auto.commit.interval=100
    kafka.consumer.auto.offset.reset=latest
    kafka.consumer.topic=test
    kafka.consumer.group.id=test
    kafka.consumer.concurrency=10
    
    kafka.producer.servers=10.93.21.21:9092
    kafka.producer.retries=0
    kafka.producer.batch.size=4096
    kafka.producer.linger=1
    kafka.producer.buffer.memory=40960

***2、Configuration：Kafka producer ***

*  1）通过@Configuration、@EnableKafka，声明Config并且打开KafkaTemplate能力。
 *  2）通过@Value注入application.properties配置文件中的kafka配置。
 *  3）生成bean，@Bean

@Configuration
    @EnableKafka
    public class KafkaProducerConfig {
    
        @Value("${kafka.producer.servers}")
        private String servers;
        @Value("${kafka.producer.retries}")
        private int retries;
        @Value("${kafka.producer.batch.size}")
        private int batchSize;
        @Value("${kafka.producer.linger}")
        private int linger;
        @Value("${kafka.producer.buffer.memory}")
        private int bufferMemory;
    
    
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            props.put(ProducerConfig.RETRIES_CONFIG, retries);
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
            props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return props;
        }
    
        public ProducerFactory<String, String> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }
    
        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() {
            return new KafkaTemplate<String, String>(producerFactory());
        }
    }

实验我们的producer，写一个Controller。想topic=test，key=key，发送消息message

@RestController
    @RequestMapping("/kafka")
    public class CollectController {
        protected final Logger logger = LoggerFactory.getLogger(this.getClass());
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        @RequestMapping(value = "/send", method = RequestMethod.GET)
        public Response sendKafka(HttpServletRequest request, HttpServletResponse response) {
            try {
                String message = request.getParameter("message");
                logger.info("kafka的消息={}"， message);
                kafkaTemplate.send("test", "key", message);
                logger.info("发送kafka成功.");
                return new Response(ResultCode.SUCCESS, "发送kafka成功", null);
            } catch (Exception e) {
                logger.error("发送kafka失败", e);
                return new Response(ResultCode.EXCEPTION, "发送kafka失败", null);
            }
        }
    
    }

***3、configuration：kafka consumer***

*  1）通过@Configuration、@EnableKafka，声明Config并且打开KafkaTemplate能力。
 *  2）通过@Value注入application.properties配置文件中的kafka配置。
 *  3）生成bean，@Bean

@Configuration
    @EnableKafka
    public class KafkaConsumerConfig {
    
        @Value("${kafka.consumer.servers}")
        private String servers;
        @Value("${kafka.consumer.enable.auto.commit}")
        private boolean enableAutoCommit;
        @Value("${kafka.consumer.session.timeout}")
        private String sessionTimeout;
        @Value("${kafka.consumer.auto.commit.interval}")
        private String autoCommitInterval;
        @Value("${kafka.consumer.group.id}")
        private String groupId;
        @Value("${kafka.consumer.auto.offset.reset}")
        private String autoOffsetReset;
        @Value("${kafka.consumer.concurrency}")
        private int concurrency;
        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            factory.setConcurrency(concurrency);
            factory.getContainerProperties().setPollTimeout(1500);
            return factory;
        }
    
        public ConsumerFactory<String, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerConfigs());
        }
    
    
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> propsMap = new HashMap<>();
            propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
            propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
            propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
            propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
            return propsMap;
        }
    
        @Bean
        public Listener listener() {
            return new Listener();
        }
    
    }

new Listener()生成一个bean用来处理从kafka读取的数据。Listener简单的实现demo如下：只是简单的读取并打印key和message值

**@KafkaListener**中topics属性用于指定kafka topic名称，topic名称由消息生产者指定，也就是由kafkaTemplate在发送消息时指定。

public class Listener {
        protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    
    
        @KafkaListener(topics = {"test"})
        public void listen(ConsumerRecord<?, ?> record) {
            logger.info("kafka的key: " + record.key());
            logger.info("kafka的value: " + record.value().toString());
        }
    }

## **kafka配置文件server.propeities文件** ##

0.11.0版本

############################# Server Basics #############################
    
    # The id of the broker. This must be set to a unique integer for each broker.
    # 节点的ID，必须与其它节点不同
    broker.id=0
    
    # Switch to enable topic deletion or not, default value is false
    # 选择启用删除主题功能，默认false
    #delete.topic.enable=true
    
    ############################# Socket Server Settings #############################
    
    # The address the socket server listens on. It will get the value returned from 
    # java.net.InetAddress.getCanonicalHostName() if not configured.
    # 套接字服务器坚挺的地址。如果没有配置，就使用java.net.InetAddress.getCanonicalHostName()的返回值
    #   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().
    # 节点的主机名会通知给生产者和消费者。如果没有设置，如果配置了"listeners"就使用"listeners"的值。
    # 否则就使用java.net.InetAddress.getCanonicalHostName()的返回值
    #advertised.listeners=PLAINTEXT://your.host.name: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
    # 服务器用来处理请求的线程数，可能包括磁盘IO
    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 seperated list of directories under which to store log files
    # 一个逗号分隔的目录列表，用来存储日志文件
    log.dirs=/tmp/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 for to ensure availability such as 3.
    # 对于除了开发测试之外的其他任何东西，group元数据内部主题的复制因子“__consumer_offsets”和“__transaction_state”，建议值大于1，以确保可用性(如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.
    # 消息直接被写入文件系统，但是默认情况下我们仅仅调用fsync()以延迟的同步系统缓存
    # 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 exceessive seeks.
    #    1. 持久性:如果不使用复制，未刷新的数据可能会丢失。
    #    2. 延迟:非常大的刷新间隔可能会在刷新时导致延迟，因为将会有大量数据刷新。
    #    3. 吞吐量:刷新通常是最昂贵的操作，而一个小的刷新间隔可能会导致过多的搜索。
    # 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.
    # 下面的设置允许你去配置刷新策略，每隔一段时间刷新或者一次N个消息（或者两个都配置）。这可以在全局范围内完成，并在每个主题的基础上重写。
    
    # 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
    # 一个日志的最小存活时间，可以被删除
    log.retention.hours=168
    
    # A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
    # segments don't drop below log.retention.bytes.
    # 一个基于大小的日志保留策略。段将被从日志中删除只要剩下的部分段不低于log.retention.bytes。
    #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).
    # Zookeeper连接字符串（具体见Zookeeper文档）
    # This is a comma separated host:port pairs, each corresponding to a zk
    # 这是一个以逗号为分割的部分，每一个都匹配一个Zookeeper
    # 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.
    # 您还可以将一个可选的chroot字符串附加到url，以指定所有kafka znode的根目录。
    zookeeper.connect=localhost:2181
    
    # Timeout in ms for connecting to zookeeper
    # 连接到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

[kafka]: https://blog.csdn.net/JinXYan/article/details/89741071