在java代码中使用kafka(springboot整合kafka)

首先，项目是个springboot-maven项目。（使用quickstart就可以）。

引入maven依赖：

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

如果是普通maven项目，也可以用这个依赖：

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.10</artifactId>
        <version>0.8.0</version>
    </dependency>

接下来，在bootstrap.yaml中添加连接kafka的配置：

spring:
  kafka:
    bootstrap-servers: 192.168.76.4:9092
    producer: # 生产者
      retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default-group
      enable-auto-commit: false
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
      # RECORD
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
      # BATCH
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
      # TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
      # COUNT
      # TIME |　COUNT　有一个条件满足时提交
      # COUNT_TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # MANUAL
      # 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
      # MANUAL_IMMEDIATE
      ack-mode: manual_immediate

也可以不在bootstrap.yaml中配置，使用配置类进行配置：

private KafkaProducetest() {
    Properties props = new Properties();
    //此处配置的是kafka的端口
    props.put("metadata.broker.list", "192.168.76.4:9092");
    
    //配置value的序列化类
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    //配置key的序列化类
    props.put("key.serializer.class", "kafka.serializer.StringEncoder");
    
    //0表示不确认主服务器是否收到消息,马上返回,低延迟但最弱的持久性,数据可能会丢失
    //1表示确认主服务器收到消息后才返回,持久性稍强,可是如果主服务器死掉,从服务器数据尚未同步,数据可能会丢失
    //-1表示确认所有服务器都收到数据,完美!
    props.put("request.required.acks", "-1");
    
    //异步生产,批量存入缓存后再发到服务器去
    props.put("producer.type", "async");
    
    //填充配置,初始化生产者
    producer = new Producer<String, String>(new ProducerConfig(props));
}

附上一些kafka的主要配置信息及默认值，仅供参考，按需使用。

broker的配置

name默认值描述brokeridnone每一个boker都有一个唯一的id作为它们的名字。 这就允许boker切换到别的主机/端口上， consumer依然知道enable.zookeeperTRUE允许注册到zookeeperlog.flush.interval.messagesLong.MaxValue在数据被写入到硬盘和消费者可用前最大累积的消息的数量log.flush.interval.msLong.MaxValue在数据被写入到硬盘前的最大时间log.flush.scheduler.interval.msLong.MaxValue检查数据是否要写入到硬盘的时间间隔。log.retention.hours168控制一个log保留多长个小时log.retention.bytes-1控制log文件最大尺寸log.cleaner.enableFALSE是否log cleaninglog.cleanup.policydeletedelete还是compat. 其它控制参数还包括log.cleaner.threads，log.cleaner.io.max.bytes.per.second，log.cleaner.dedupe.buffer.size，log.cleaner.io.buffer.size，log.cleaner.io.buffer.load.factor，log.cleaner.backoff.ms，log.cleaner.min.cleanable.ratio，log.cleaner.delete.retention.mslog.dir/tmp/kafka-logs指定log文件的根目录log.segment.bytes1024*1024单一的log segment文件大小log.roll.hours24 * 7开始一个新的log文件片段的最大时间message.max.bytes1000000 + MessageSet.LogOverhead一个socket 请求的最大字节数num.network.threads3处理网络请求的线程数num.io.threads8处理IO的线程数background.threads10后台线程序num.partitions1默认分区数socket.send.buffer.bytes102400socket SO_SNDBUFF参数socket.receive.buffer.bytes102400socket SO_RCVBUFF参数zookeeper.connectlocalhost:2182/kafka指定zookeeper连接字符串， 格式如hostname:port/chroot。chroot是一个namespace

consumer的配置

name默认值描述groupidgroupid一个字符串用来指示一组consumer所在的组socket.timeout.ms30000socket超时时间socket.buffersize64*1024socket receive bufferfetch.size300 * 1024控制在一个请求中获取的消息的字节数。 这个参数在0.8.x中由fetch.message.max.bytes,fetch.min.bytes取代backoff.increment.ms1000这个参数避免在没有新数据的情况下重复频繁的拉数据。 如果拉到空数据，则多推后这个时间queued.max.message.chunks2high level consumer内部缓存拉回来的消息到一个队列中。 这个值控制这个队列的大小autocommit.enableTRUE如果true,consumer定期地往zookeeper写入每个分区的offsetauto.commit.interval.ms10000往zookeeper上写offset的频率auto.offset.resetsmallnest如果offset出了返回，则 smallest : 自动设置reset到最小的offset. largest : 自动设置offset到最大的offset. anything else : 否则抛出异常.consumer.timeout.ms-1默认-1,consumer在没有新消息时无限期的block。如果设置一个正值， 一个超时异常会抛出rebalance.retries.max4rebalance时的最大尝试次数

producer的配置

name默认值描述serializer.classkafka.serializer.DefaultEncoder必须实现kafka.serializer.Encoder 接口，将T类型的对象encode成kafka messagekey.serializer.classserializer.classkey对象的serializer类partitioner.classkafka.producer.DefaultPartitioner必须实现kafka.producer.Partitioner ，根据Key提供一个分区策略producer.typesync指定消息发送是同步还是异步。异步asyc成批发送用kafka.producer.AyncProducer， 同步sync用kafka.producer.SyncProducermetadata.broker.listboker list使用这个参数传入boker和分区的静态信息，如host1:port1,host2:port2, 这个可以是全部boker的一部分compression.codecNoCompressionCodec消息压缩，默认不压缩compressed.topicsnull在设置了压缩的情况下，可以指定特定的topic压缩，为指定则全部压缩message.send.max.retries3消息发送最大尝试次数retry.backoff.ms300每次尝试增加的额外的间隔时间topic.metadata.refresh.interval.ms600000定期的获取元数据的时间。当分区丢失，leader不可用时producer也会主动获取元数据，如果为0，则每次发送完消息就获取元数据，不推荐。如果为负值，则只有在失败的情况下获取元数据。queue.buffering.max.ms5000在producer queue的缓存的数据最大时间，仅仅for asycqueue.buffering.max.message10000producer 缓存的消息的最大数量，仅仅for asycqueue.enqueue.timeout.ms-10当queue满时丢掉，负值是queue满时block,正值是queue满时block相应的时间，仅仅for asycbatch.num.messages200一批消息的数量，仅仅for asyc
有了配置之后，就可以开始写生产者代码：

    @Autowired
    //引入kafka的template
    private KafkaTemplate<String,String> kafkaTemplate;
    //如果是配置类，就注入配置类就可以

    //在controller中直接进行消息的发布(为了简便，不建议)
    @GetMapping("/kafka")
    public boolean kafka(@RequestParam("msg") String msg){
        kafkaTemplate.send("test", msg);
        return true;
    }

在linux命令行输入如下指令即可打开消费者控制台，查看消息：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

启动服务，调用接口发布消息：

在linux界面接收到，说明连接成功。

在服务器本地可以使用命令行参数连接并且进行发布订阅操作，但是使用spring boot连接服务器时可能会有如下错误：

Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available

这个时候，停掉kafka，打开kafka的配置文件(server.properties)，修改以下两个地方，默认第二个红框的内容是注释掉的，这样的话虚拟机之外的是访问不进来的，只需要将注释放开，改成自己的ip:端口即可。

接下来，咱们完成一下消费者代码：

@Component
public class Consumer {
    
    /**
     * @param record ack
     * @KafkaListener(topics = "",groupId = "")
     * @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
     * @TopicPartition(topic = "topic2", partitions = "0",
     * partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
     * },concurrency = "6")
     * //concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
     */

    //因为在配置中已经指令了groupId,故无需再次指定,可以根据自己的需求增加属性
    @KafkaListener(topics = "test")
    public void consumer1(ConsumerRecord<String,String> record, Acknowledgment ack){
        //简单打印一下收到的消息
        System.out.println(record);
        //手动提交offset，一般是提交一个banch，幂等性防止重复消息
        ack.acknowledge();
    }
}

消息消费成功。

ConsumerRecord(topic = test, partition = 0, leaderEpoch = 0, offset = 17, CreateTime = 1662430642575, serialized key size = -1, serialized value size = 5, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = kafka)

至此，生产者和消费者的简单代码完成，当然了，现实可能会遇到更多复杂的生产和消费的问题，这个就需要自己去探索了~