kafka编程应用
2020-04-22 本文已影响0人
北海北_6dc3
Kafka中的分区器、序列化器、拦截器是否了解？它们之间的处理顺序是什么？
- 序列化器：生产者需要用序列化器（Serializer）把对象转换成字节数组才能通过网络发送给 Kafka。而在对侧，消费者需要用反序列化器（Deserializer）把从 Kafka 中收到的字节数组转换成相应的对象。
- 分区器：分区器的作用就是为消息分配分区。如果消息 ProducerRecord 中没有指定 partition 字段，那么就需要依赖分区器，根据 key 这个字段来计算 partition 的值。
- Kafka 一共有两种拦截器：生产者拦截器和消费者拦截器。
1、生产者拦截器既可以用来在消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息、修改消息的内容等，也可以用来在发送回调逻辑前做一些定制化的需求，比如统计类工作。
消费者拦截器主要在消费到消息或在提交消费位移时进行一些定制化的操作。
2、消息在通过 send() 方法发往 broker 的过程中，有可能需要经过拦截器（Interceptor）、序列化器（Serializer）和分区器（Partitioner）的一系列作用之后才能被真正地发往 broker。拦截器（下一章会详细介绍）一般不是必需的，而序列化器是必需的。消息经过序列化之后就需要确定它发往的分区，如果消息 ProducerRecord 中指定了 partition 字段，那么就不需要分区器的作用，因为 partition 代表的就是所要发往的分区号。
处理顺序 ：拦截器->序列化器->分区器
KafkaProducer 在将消息序列化和计算分区之前会调用生产者拦截器的 onSend() 方法来对消息进行相应的定制化操作。
然后生产者需要用序列化器（Serializer）把对象转换成字节数组才能通过网络发送给 Kafka。
最后可能会被发往分区器为消息分配分区。
kafka 远程连接
首先，我们需要下载与服务端相同的kafka版本地址
https://downloads.apache.org/kafka/2.4.0/kafka_2.12-2.4.0.tgz
服务端需要监听外网ip,需要暴露的外网
advertised.listeners=PLAINTEXT://xxx:9092
执行命令
D:\Learn Study\springBoot\springBoot\simple\kafka>cd D:\opt\kafka_2.12-2.4.0\bin\windows
D:\opt\kafka_2.12-2.4.0\bin\windows>kafka-topics.bat --create --zookeeper 47.105.194.139:2181 --replication-factor 1 --partitions 1 --topic test5
Picked up _JAVA_OPTIONS: -Djava.net.preferIPv4Stack=true
Created topic test5.
说明已经成功连接了。
我们可以基于程序编程了。
1、建立项目，引入pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.iva.study</groupId>
<artifactId>spring-boot</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>spring-boot-kafka</name>
<description>spring-boot-kafka</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<!--核心依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<!--核心依赖End-->
<!--辅助：可以修改不重启-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<!--可以修改不重启End-->
<!--测试-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
<!--测试End-->
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2、添加配置
spring.kafka.bootstrap-servers=47.105.194.139:9092
3、开始编程
@SpringBootApplication
public class SpringBootKafkaApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(SpringBootKafkaApplication.class, args);
}
private final Logger logger = LoggerFactory.getLogger(SpringBootKafkaApplication.class);
@Autowired
private KafkaTemplate<Object, Object> template;
@KafkaListener(id = "webGroup", topics = "test5")
public void listen(String input) {
logger.info("input value: {}" , input);
}
@Override
public void run(String... strings) throws Exception {
this.template.send("test5", "haha");
}
}
Spring-kafka-test嵌入式Kafka Server
不过上面的代码能够启动成功，前提是你已经有了Kafka Server的服务环境，我们知道Kafka是由Scala + Zookeeper构建的，可以从官网下载部署包在本地部署。但是，我想告诉你，为了简化开发环节验证Kafka相关功能，Spring-Kafka-Test已经封装了Kafka-test提供了注解式的一键开启Kafka Server的功能，使用起来也是超级简单。本文后面的所有测试用例的Kafka都是使用这种嵌入式服务提供的。
创建topic方法
- bean自动创建【不方便，需要编码】
@Configuration
public class KafkaInitialConfiguration {
//创建TopicName为topic.quick.initial的Topic并设置分区数为8以及副本数为1
@Bean
public NewTopic initialTopic() {
return new NewTopic("topic.quick.initial",8, (short) 1 );
}
}
- bean手动创建
暴露对象想AdminClient
@Configuration
public class KafkaInitialConfiguration1 {
@Autowired
KafkaAdmin kafkaAdmin;
@Bean
public AdminClient adminClient() {
return AdminClient.create(kafkaAdmin.getConfig());
}
}
调用
@SpringBootApplication
public class SpringBootKafkaApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(SpringBootKafkaApplication.class, args);
}
@Autowired
private AdminClient adminClient;
@Override
public void run(String... strings) throws Exception {
NewTopic topic = new NewTopic("topic7",2, (short) 1 );
adminClient.createTopics(Arrays.asList(topic));
}
}
查询topic信息
@Override
public void run(String... strings) throws Exception {
DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList("test"));
result.all().get().forEach((k,v)->System.out.println("k: "+k+" ,v: "+v.toString()+"\n"));
}
kafka生产者
- 异步
@Override
public void run(String... strings) throws Exception {
template.send("test5","k2").addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {
@Override
public void onFailure(Throwable throwable) {
System.out.println("kafka执行失败");
}
@Override
public void onSuccess(SendResult<Object, Object> objectObjectSendResult) {
System.out.println("kafka执行成功"+objectObjectSendResult.getProducerRecord().value());
}
});
}
- 同步
@Override
public void run(String... strings) throws Exception {
ListenableFuture<SendResult<Object,Object>> future = template.send("test5","kl");
try {
SendResult<Object,Object> result = future.get();
System.out.println(result.getProducerRecord().value());
}catch (Throwable e){
e.printStackTrace();
}
}
kafka消费者
@KafkaListener(id = "webGroup", topics = "test5")
public void listen(String input) {
logger.info("input value: {}" , input);
}
KAFKA事务消息
默认情况下，Spring-kafka自动生成的KafkaTemplate实例，是不具有事务消息发送能力的。需要使用如下配置激活事务特性。事务激活后，所有的消息发送只能在发生事务的方法内执行了，不然就会抛一个没有事务交易的异常
spring.kafka.producer.transaction-id-prefix=kafka_tx.
当发送消息有事务要求时，比如，当所有消息发送成功才算成功，如下面的例子：假设第一条消费发送后，在发第二条消息前出现了异常，那么第一条已经发送的消息也会回滚。而且正常情况下，假设在消息一发送后休眠一段时间，在发送第二条消息，消费端也只有在事务方法执行完成后才会接收到消息
- 方式一，不用注解
@Override
public void run(String... strings) throws Exception {
template.executeInTransaction(t ->{
t.send("test5","k4");
if("error".equals("error")){
throw new RuntimeException("failed");
}
t.send("test5","ckl");
return true;
});
}
- 方式二，注解
@Transactional(rollbackFor = RuntimeException.class)
@Override
public void run(String... strings) throws Exception {
template.send("test5","k4");
if("error".equals("error")){
throw new RuntimeException("failed");
}
template.send("test5","ckl");
}
参考资料：
springboot集成kafka示例
spring boot集成kafka之spring-kafka深入探秘
Spring-Kafka（三）—— 操作Topic以及Kafka Tool 2的使用
