kafka接收数据

（注: 这是在已搭建好的kafka中进行的，如果有kafka的IP地址你不用在自己电脑上搭建）

第一步 导入jar

如果你在发送数据那已经添加了kafka的jar在这就不用加了（不同系统需要，别发神经以为不要）

       <!-- 这是kafka的jar包  -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        
        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <!-- json -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>

第二步 配置myl

spring:
  kafka:
    consumer:
      # 指定kafka server的地址，集群配多个，中间，逗号隔开
      bootstrap-servers: localhost
      #如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
      auto-commit-interval: 100
      # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
      auto-offset-reset: earliest
      #--> 设置自动提交offset
      enable-auto-commit: true
      # 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
      group-id: bayerConsumerGroup
      # 指定消息key和消息体的编解码方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

第三步 接受数据

package com.jianxun.nash.consumer.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.jianxun.nash.consumer.model.TopicMessageDTO;
import com.jianxun.nash.consumer.model.hisModel.SurgeryOrder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class ConsumerListener {
      
    /**
     *  消费 
     *  his_hospital_1 是获取数据的标识，可在我kafka的发送博客中查看
     *  topics 里的数据在 我kafka的发送博客中  
     *  TopicMessageDTO 在 kafka的发送博客中也有提到。不同的系统要把这个类复制过来 
     *  SurgeryOrder你的数据对象而已 自己定义的要跟发送的数据一直
     * @param record
     */
    @KafkaListener(topics = "his_hospital_1")
    public void consumerHisData(ConsumerRecord<String, String> record){
        // 将 json转换数据
        TopicMessageDTO<List<SurgeryOrder>> listTopicMessageDTO = 
                JSON.parseObject(record.value(), new TypeReference<TopicMessageDTO<List<SurgeryOrder>>>() {});
    }

}

发送数据

kafka发送数据博客