docker-kafka情况搭建

拉取镜像

下载zookeeper

• 1. docker pull wurstmeister/zookeeper

  复制代码

下载kafka

• 1. docker pull wurstmeister/kafka

  复制代码

启动镜像

启动zookeeper

• 1. docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

  复制代码

启动kafka

• 1. docker run -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=192.168.3.200:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.3.200:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

  复制代码
• 此中ip地址需换成宿主机的ip地址
  

测试kafka

进入kafka容器

• 1. docker exec -it kafka /bin/bash

  复制代码

进入kafka所在目次

• 1. cd /opt/kafka/

  复制代码

再开一个xshell客户端，一个作为生产者，一个作为消费者，按上面操作进入kafka目次

发送消息

• 1. ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka

  复制代码

接收消息

• 1. ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka --from-beginning

  复制代码

检察topic

• 1. ./bin/kafka-topics.sh --zookeeper 192.168.3.200:2181 --list

  复制代码

springboot-kafka简朴demo

pom.xml

• 1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-web</artifactId>
  4. </dependency>
  5. <!--spring-kafka-->
  6. <dependency>
  7.     <groupId>org.springframework.kafka</groupId>
  8.     <artifactId>spring-kafka</artifactId>
  9.     <version>2.4.4.RELEASE</version>
  10. </dependency>
  11. <dependency>
  12.     <groupId>org.projectlombok</groupId>
  13.     <artifactId>lombok</artifactId>
  14. </dependency>

  复制代码

问题记录

• 1. org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration': Unexpected exception during bean creation; nested exception is java.lang.TypeNotPresentException: Type org.springframework.kafka.listener.RecordInterceptor not present
  2.     at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:544) ~[spring-beans-5.3.2.jar:5.3.2]
  3.     at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:335) ~[spring-beans-5.3.2.jar:5.3.2]
  4.     at org.springframework.beans.factory.support.AbstractBeanFactory$$Lambda$197/106557175.getObject(Unknown Source) ~[na:na]

  复制代码
• 问题原因：springboot版本与spring-kafka版本不对应，项目中使用的springboot版本是：2.4.1，颠末测试，spring-kafka的版本需要在2.4.4.RELEASE以上，否则会报上面的非常
  
• springboot版本与spring-kafka版本的对应关系，可以检察官网
  
• 下面是官网截图：
  
• 
  
  

application.yml

• 1. spring:
  2.   kafka:
  3.     producer:
  4.       bootstrap-servers: 192.168.3.200:9092
  5.     consumer:
  6.       bootstrap-servers: 192.168.3.200:9092
  7.       group-id: 0

  复制代码

KafkaController.java

• 1. @RestController
  2. public class KafkaController {
  4.     @Resource
  5.     private KafkaProducer kafkaProducer;
  7.     @GetMapping("/sendMessage")
  8.     public String sendMessage(String message) {
  9.         kafkaProducer.sendMessage(message);
  10.         return "success";
  11.     }
  12. }

  复制代码

KafkaProducer.java

• 1. @Slf4j
  2. @Component
  3. public class KafkaProducer {
  5.     @Resource
  6.     private KafkaTemplate<String, String> kafkaTemplate;
  8.     public void sendMessage(String message) {
  9.         ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("kafka", message);
  10.         future.addCallback(o -> log.info("kafka-消息发送成功：" + message), throwable -> log.info("kafka-消息发送失败：" + message));
  11.     }
  12. }

  复制代码

KafkaConsumer.java

• 1. @Slf4j
  2. @Component
  3. public class KafkaConsumer {
  5.     @KafkaListener(topics = {"kafka"})
  6.     public void listener(String message){
  7.         log.info("kafka-收到消息："+message);
  8.     }
  9. }

  复制代码
• 此时已经可以启动项目，访问：http://localhost:8080/sendMessage?message=abc 进行测试
  

整合swagger-ui

pom.xml

• 1. <!--swagger-ui-->
  2.         <dependency>
  3.             <groupId>io.springfox</groupId>
  4.             <artifactId>springfox-swagger2</artifactId>
  5.             <version>2.9.2</version>
  6.         </dependency>
  7.                 <dependency>
  8.             <groupId>io.springfox</groupId>
  9.             <artifactId>springfox-swagger-ui</artifactId>
  10.             <version>2.9.2</version>
  11.         </dependency>
  12.         <!--增强swagger-ui-->
  13.         <dependency>
  14.             <groupId>com.github.xiaoymin</groupId>
  15.             <artifactId>knife4j-spring-boot-starter</artifactId>
  16.             <version>2.0.4</version>
  17.         </dependency>

  复制代码

Swagger2Config.java

• 1. @Configuration
  2. @EnableSwagger2
  3. @EnableKnife4j
  4. public class Swagger2Config {
  6.     @Bean
  7.     public Docket createRestApi() {
  8.         return new Docket(DocumentationType.SWAGGER_2)
  9.                 .apiInfo(apiInfo())
  10.                 .select()
  11.                 .apis(RequestHandlerSelectors.basePackage("com.example.kafka"))
  12.                 .paths(PathSelectors.any())
  13.                 .build();
  14.     }
  16.     private ApiInfo apiInfo() {
  17.         return new ApiInfoBuilder().title("DEMO--API接口文档")
  18.                 .description("接口返回数据格式为json。其基本结构为{'code': 200, 'msg':'', 'data':{...} }。其中code为错误代码，我们采用"
  19.                         + "标准的http错误代码，比如200代表成功，400代表参数错误，401代表没有登录，500表示服务器内部错误无法完成请求"
  20.                         + "。msg为错误消息，大部分时候为空，当code代码不为200时会是具体的错误描述。data" + "是返回的数据，data的具体内容请参考api。")
  21.                 .termsOfServiceUrl("http://www.xxx.com/")
  22.                 .version("1.0")
  23.                 .build();
  24.     }
  25. }

  复制代码

测试地址

• 原生swagger-ui：http://localhost:8080/swagger-ui.html
  
• 增强swagger-ui：http://localhost:8080/doc.html
  
• 增强swagger-ui界面比原生好看一些（尚有一些新增功能，可以百度相识），使用也比较简朴，只需引入knife4j-spring-boot-starter依赖，在swagger配置类上加@EnableKnife4j注解就可以了
  

kafka相干概念学习

topic-partition

• Topic 是一个存储消息的逻辑概念，可以明白为是一个消息的集合
  
• Partition，分区，提高了Kafka的并发，也解决了Topic中数据的负载均衡
  
• Kafka 只保证在同一个分区内的消息是有序的
  

consumer group

• 消费者组，由多个 consumer 组成， 消费者组内每个消费者负 责消费不同分区的数据，一个分区只能由一个组内的某一个消费者消费；消费者组之间互不影响
  
• 消费者组是逻辑上的一个订阅者
  

producer消息分发策略

• 在 Kafka 中，一条消息由 key 和 value 两部门组成
  
• producer 则会根据 key 和 partition 机制，来判定当前这条消息应该发送并存储到哪个 partition 中
  
• Kafka 采用的是 hash 取模的分区算法
  
• 如果 key 为 null 的话，则会随机的分配一个分区，这个随机是在这个参数 "metadata.max.age.ms"的时间范围内随机选择一个
  

消费者如何消费指定分区消息

• 1. /**
  2. * 1.使用 spring-kafka.jar包中的 KafkaTemplate 类型
  3. *   使用 @KafkaListener 注解方式
  4. *   如下：说明消费的是名称为test的topic下,分区 1 中的消息
  5. */
  6. @KafkaListener(topicPartitions = {@TopicPartition(topic = "test",partitions = {"1"})})
  8. /**
  9. * 2.使用kafka-clients.jar包中的 KafkaConsumer 类型
  10. * 如下：说明消费的是名称为test的topic下,分区 1 中的消息
  11. */
  12. TopicPartition topicPartition = new TopicPartition("test" , 1);
  13. KafkaConsumer consumer = new KafkaConsumer(props);
  14. consumer.assign(Arrays.asList(topicPartition));

  复制代码

分区分配策略

多个Partitions有什么好处

• 多个 partition ，能够对 broker 上的数据进行分片，通过淘汰消息容量来提升 IO 性能
  
• 为了提高消费端的消费本领，一般情况下会通过多个 conusmer 去消费 同一个 topic 中的消息，即实现消费端的负载均衡
  

针对多个Partition，消费者该消费哪个分区的消息

• RangeAssignor 分配策略(范围分区)
  
  
  
  • 默认采用 Range 范围分区
    
  • 通过 partitions数/consumer数 来决定每个消费者应该消费几个分区
    
  
  
• RoundRobinAssignor分配策略(轮询分区)
  

什么时间触发分区分配策略（Kafka 消费者端的 Rebalance 操作）

• 同一个 consumer 消费者组 group.id 中，新增了消费者进来，会执行 Rebalance 操作
  
• 消费者离开当期所属的 consumer group组。比如 主动停机 或者 宕机
  
• 分区数量发生变化时(即 topic 的分区数量发生变化时)
  
• 消费者主动取消订阅
  

Rebalance，即对 Kafka 中的分区进行重新分配的过程

• Rebalance，即对 Kafka 中的分区进行重新分配的过程
  

确定 Coordinator

• 当消费者向 kafka 集群中的任意一个 broker 发送一个 GroupCoordinatorRequest 请求，Kafka Server 服务端会返回一个当前负载最小的 broker 节点的 id，并将该 id 所对应的的 broker 节点设置为当前 consumer group 的 Coordinator
  

JoinGroup 过程

• 当确定了 Coordinator 之后，所有的 Consumer 都会向 Coordinator 发送一个 JoinGroup 请求（只要启动，所有消费者都会发送该请求），此时 Coordinator 会从 Consumer group 中选取一个 consumer 担任 leader 脚色，并把组成员信息和订阅的消息发送给所有的消费组
  

Synchronizing Group State 阶段

• 该阶段主要完成 leader 将消费者对应的 partition 分配方案同步给consumer group 中的所有 consumer
  
• 每个消费者，都会向 Coordinator 发送 SyncGroupRequest 请求
  
• 每个消费者，还会向 Coordinator 发送 SyncGroup 请求，不过只有 leader 节点会发送分配方案
  
• 当 leader 把方案发给 Coordinator 以后，Coordinator 会把结果设置到 SyncGroupResponse 中，如许所有成员都知道本身应该消费哪个分区
  

零拷贝

• 使用DMA技术，通过网卡直接去访问系统的内存
  
• 去掉那些没必要的数据复制操作， 同时也会淘汰上下文切换次数
  
• Java 提供了访问这个系统调用的方法：FileChannel.transferTo API
  
• linux 系统调用sendfile()
  
• 使用 sendfile() ，只需要一次拷贝就行，允许操作系统将数据直接从页缓存发送到网络上，以是在这个优化的路径中， 只有最后一步将数据拷贝到网卡缓存中是需要的
  

java零拷贝实现

• 1. File file = new File("demo.zip");
  2. RandomAccessFile raf = new RandomAccessFile(file, "rw");
  3. FileChannel fileChannel = raf.getChannel();
  4. SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("", 1234));
  5. // 直接使用了transferTo()进行通道间的数据传输
  6. fileChannel.transferTo(0, fileChannel.size(), socketChannel);

  复制代码

学习链接


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。