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

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

引入maven依赖：




<dependency>



    <groupId></groupId>



    <artifactId>spring-kafka</artifactId>



 </dependency>

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




    <dependency>



        <groupId></groupId>



        <artifactId>kafka_2.10</artifactId>



        <version>0.8.0</version>



    </dependency>

接下来，在中添加连接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: 



      value-serializer: 



    consumer:



      group-id: default-group



      enable-auto-commit: false




      auto-offset-reset: earliest



      key-deserializer: 



      value-deserializer: 



    listener:



      # 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交



      # RECORD



      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交



      # BATCH



      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交



      # TIME



      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交



      # COUNT



      # TIME |　COUNT　有一个条件满足时提交



      # COUNT_TIME



      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用()后提交



      # MANUAL



      # 手动调用()后立即提交，一般使用这种



      # MANUAL_IMMEDIATE



      ack-mode: manual_immediate

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





private KafkaProducetest() {



    Properties props = new Properties();



    //此处配置的是kafka的端口




    ("", "192.168.76.4:9092");



    



    //配置value的序列化类




    ("", "");



    //配置key的序列化类




    ("", "");



    



    //0表示不确认主服务器是否收到消息,马上返回,低延迟但最弱的持久性,数据可能会丢失




    //1表示确认主服务器收到消息后才返回,持久性稍强,可是如果主服务器死掉,从服务器数据尚未同步,数据可能会丢失




    //-1表示确认所有服务器都收到数据,完美!




    ("", "-1");



    



    //异步生产,批量存入缓存后再发到服务器去




    ("", "async");



    



    //填充配置,初始化生产者




    producer = new Producer<String, String>(new ProducerConfig(props));



}

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

broker的配置

name	默认值	描述
brokerid	none	每一个boker都有一个唯一的id作为它们的名字。这就允许boker切换到别的主机/端口上， consumer依然知道
	TRUE	允许注册到zookeeper
		在数据被写入到硬盘和消费者可用前最大累积的消息的数量
		在数据被写入到硬盘前的最大时间
		检查数据是否要写入到硬盘的时间间隔。
	168	控制一个log保留多长个小时
	-1	控制log文件最大尺寸
	FALSE	是否log cleaning
	delete	delete还是compat. 其它控制参数还包括，，，，，，，
	/tmp/kafka-logs	指定log文件的根目录
	1024*1024	单一的log segment文件大小
	24 * 7	开始一个新的log文件片段的最大时间
	1000000 +	一个socket 请求的最大字节数
	3	处理网络请求的线程数
	8	处理IO的线程数
	10	后台线程序
	1	默认分区数
	102400	socket SO_SNDBUFF参数
	102400	socket SO_RCVBUFF参数
	localhost:2182/kafka	指定zookeeper连接字符串，格式如hostname:port/chroot。chroot是一个namespace

consumer的配置

name	默认值	描述
groupid	groupid	一个字符串用来指示一组consumer所在的组
	30000	socket超时时间
	64*1024	socket receive buffer
	300 * 1024	控制在一个请求中获取的消息的字节数。这个参数在0.中由,取代
	1000	这个参数避免在没有新数据的情况下重复频繁的拉数据。如果拉到空数据，则多推后这个时间
	2	high level consumer内部缓存拉回来的消息到一个队列中。这个值控制这个队列的大小
	TRUE	如果true,consumer定期地往zookeeper写入每个分区的offset
	10000	往zookeeper上写offset的频率
	smallnest	如果offset出了返回，则 smallest : 自动设置reset到最小的offset. largest : 自动设置offset到最大的offset. anything else : 否则抛出异常.
	-1	默认-1,consumer在没有新消息时无限期的block。如果设置一个正值，一个超时异常会抛出
	4	rebalance时的最大尝试次数

producer的配置

name	默认值	描述
		必须实现接口，将T类型的对象encode成kafka message
		key对象的serializer类
		必须实现，根据Key提供一个分区策略
	sync	指定消息发送是同步还是异步。异步asyc成批发送用，同步sync用
	boker list	使用这个参数传入boker和分区的静态信息，如host1:port1,host2:port2, 这个可以是全部boker的一部分
	NoCompressionCodec	消息压缩，默认不压缩
	null	在设置了压缩的情况下，可以指定特定的topic压缩，为指定则全部压缩
	3	消息发送最大尝试次数
	300	每次尝试增加的额外的间隔时间
	600000	定期的获取元数据的时间。当分区丢失，leader不可用时producer也会主动获取元数据，如果为0，则每次发送完消息就获取元数据，不推荐。如果为负值，则只有在失败的情况下获取元数据。
	5000	在producer queue的缓存的数据最大时间，仅仅for asyc
	10000	producer 缓存的消息的最大数量，仅仅for asyc
	-1	0当queue满时丢掉，负值是queue满时block,正值是queue满时block相应的时间，仅仅for asyc
	200	一批消息的数量，仅仅for asyc

有了配置之后，就可以开始写生产者代码：




    @Autowired




    //引入kafka的template




    private KafkaTemplate<String,String> kafkaTemplate;



    //如果是配置类，就注入配置类就可以




 



    //在controller中直接进行消息的发布(为了简便，不建议)




    @GetMapping("/kafka")




    public boolean kafka(@RequestParam("msg") String msg){



        ("test", msg);



        return true;



    }

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

./ --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){



        //简单打印一下收到的消息




        (record);



        //手动提交offset，一般是提交一个banch，幂等性防止重复消息




        ();



    }



}

消息消费成功。

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)

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

秒客网

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

broker的配置

consumer的配置

producer的配置

相关文章