二丶SpringBoot整合Kafka

  • Z时代
  • 2024-01-10
  • 分类：综合

编程

 

1.parent的pom.xml配置

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.5.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
</dependencies>

 

2.下面是kafka-producer代码  

server.servlet.context-path=/producer
server.port=8001
## Spring 整合 kafka
spring.kafka.bootstrap-servers=192.168.11.51:9092
## kafka producer 发送消息失败时的一个重试的次数
spring.kafka.producer.retries=0
## 批量发送数据的配置
spring.kafka.producer.batch-size=16384
## 设置kafka 生产者内存缓存区的大小（32M）
spring.kafka.producer.buffer-memory=33554432
## kafka消息的序列化配置
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
# acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
# acks=-1: 表示分区leader必须等待消息被成功写入到所有的ISR副本(同步副本)中才认为producer请求成功。这种方案提供最高的消息持久性保证，但是理论上吞吐率也是最差的。
##     这个是kafka生产端最重要的选项
spring.kafka.producer.acks=1

 

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Application {
public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
}
}

 

package com.bfxy.kafka.producer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
 * @auther daiison
 * @create 2020-04-2020/4/27-2:27 下午
 **/
@Slf4j
@Component
public class KafkaProducerService {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
    public void sendMessage(String topic, Object object) {
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, object);
future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable throwable) {
log.error("发送消息失败: " + throwable.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
log.info("发送消息成功: " + result.toString());
}
        });
}
}

 

测试类

package com.bfxy.kafka.producer.test;
import com.bfxy.kafka.producer.KafkaProducerService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
 * @auther daiison
 * @create 2020-04-2020/4/27-3:04 下午
 **/
@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {
@Autowired
private KafkaProducerService kafkaProducerService;
@Test
public void send() throws InterruptedException {
        String topic = "topic02";
        for(int i=0; i < 1000; i ++) {
kafkaProducerService.sendMessage(topic, "hello kafka" + i);
Thread.sleep(5);
}
        Thread.sleep(Integer.MAX_VALUE);
}
}

 

3.下面是kafka-consumer代码  

server.servlet.context-path=/consumser
server.port=8002
spring.kafka.bootstrap-servers=192.168.11.51:9092
## consumer 消息的签收机制：手工签收
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.listener.ack-mode=manual
# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
# latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
# earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
spring.kafka.consumer.auto-offset-reset=earliest
## 序列化配置
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.listener.concurrency=5

 

package com.bfxy.kafka;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
 * @auther daiison
 * @create 2020-04-2020/4/27-2:56 下午
 **/
@SpringBootApplication
public class Application {
public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
}
}

 

package com.bfxy.kafka.consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
/**
 * @auther daiison
 * @create 2020-04-2020/4/27-2:58 下午
 **/
@Slf4j
@Component
public class KafkaConsumerService {
@KafkaListener(groupId = "group02",topics = "topic02")
public void onMessage(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
log.info("消费端接收消息: {}", record.value());
// 收工签收机制
acknowledgment.acknowledge();
}
}

以上是 二丶SpringBoot整合Kafka 的全部内容， 来源链接： utcz.com/z/515931.html

回到顶部