一文带您快速入门Kafka

rpm -qa|grep yum

tar -zxvf zookeeper-3.4.6.tar.gz

mkdir data 
mkdir logs

cd conf
cp zoo_sample.cfg zoo.cfg

# ZooKeeper 服务器心跳时间，单位:毫秒
tickTime=2000
# 投票选举新 Leader 的初始化时间
initLimit=10
# Leader 与 Flower 心跳检测最大容忍时间，响应超过 syncLimit*tickTime，就剔除 Flower
syncLimit=5
# 存放数据的文件夹
dataDir=/root/zookeeper-3.4.6/data
# 存放日志的文件夹
dataLogDir=/root/zookeeper-3.4.6/logs
# ZooKeeper提供给接入客户端的连接端口
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature

cat > myid
vi myid

vi /etc/profile
 export ZOOKEEPER_HOME=/root/zookeeper-3.4.6
export PATH=$PATH:$ZOOKEEPER_HOME/bin

# put dir remoteDir
put D:\downloads\kafka_2.13-3.5.0.tgz /root/kafka_2.13-3.5.0.tgz

tar -zxvf kafka_2.13-3.5.0.tgz

cd kafka_2.13-3.5.0cd config/

# 是Broker的标识，因此在集群中必需唯一
broker.id=0
# Broker 对外服务地址(我这里vmware的ip是192.168.226.140)
listeners=PLAINTEXT://192.168.226.140:9092
# 实际工作中，会分内网外网，当有需要提供给外部客户端使用时，我们一般 listeners 配置内网供 Broker 之间通信使用，而 advertised.listeners 配置走外网给接入的客户端使用
#advertised.listeners=PLAINTEXT://your.host.name:9092
# 存放消息日志文件地址
log.dirs=/root/kafka_2.13-3.5.0/logs
# ZK 的访问路径，我这里因为 ZK 和 Kafka 放在了同一个服务器上，所以就使用了 localhost
zookeeper.connect=localhost:2181

vi /etc/profile
 export KAFKA_HOME=/root/kafka_2.13-3.5.0
export PATH=$PATH:$KAFKA_HOME/bin

kafka-server-start.sh ../config/server.properties &

    4.0.0
    
        org.springframework.boot
        spring-boot-starter-parent
        
        2.7.12
         
    
    com.example.czl
    kafka
    0.0.1-SNAPSHOT
    springboot-kafka
    spring boot集成kafka demo
    
        1.8
        3.5.3.1
        2.3
        1.18.26
        31.1-jre
    
    
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.springframework.kafka
            spring-kafka
        

        
            com.mysql
            mysql-connector-j
            runtime
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        
        
            org.springframework.kafka
            spring-kafka-test
            test
            
                
                    scala-library
                    org.scala-lang
                
                
                    scala-reflect
                    org.scala-lang
                
            
        

        
        
            com.baomidou
            mybatis-plus-boot-starter
            ${mybatis-plus.version}
        
        
            com.baomidou
            mybatis-plus-generator
            ${mybatis-plus.version}
        
        
            org.apache.velocity
            velocity-engine-core
            ${velocity-engine-core.version}
        

        
            org.projectlombok
            lombok
            ${lombok.version}
        

        
            com.google.guava
            guava
            ${guava.version}
        
    

    
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
            
        
    

spring:
  application:
    name: spring-boot-kafka
  profiles:
    active: dev

server:
  port: 8080

spring:
  datasource:
    url: "jdbc:mysql:/***?useSSL=false&useUnicode=true&characterEncoding=utf8&ApplicationName=spring-boot-demo&serverTimezone=UTC&allowMultiQueries=true"
    username: "***"
    password: "***"
  kafka:
    bootstrap-servers: "192.168.226.140:9092" # 访问Kafka服务端的地址
    consumer:
      group-id: ${spring.application.name}-${spring.profiles.active} # 一条消息只会被订阅了该主题的同一个分组内的一个消费者消费

mybatis-plus:
  configuration:
    # 打印sql
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

    
    

    
    
        
            %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36}\(%L\) - [%X{traceId}] %msg%n
        
        
        
            ${LOG_PATH_HOME}/log.%d{yyyy-MM-dd}.%i.log
            200MB
        
    

    
    
        
            %d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36}\(%L\) - [%X{traceId}] %msg%n
        
    

    
    
    
    
        
        
    

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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;


@Slf4j
@Component
@RequiredArgsConstructor
public class ProducerDemo {
    private final KafkaTemplate kafkaTemplate;

    
    public void send(String topic, String msg, ListenableFutureCallback> callback) {
        log.info("发送Kafka消息 - topic : {}, msg : {}", topic, msg);
        ListenableFuture> future = kafkaTemplate.send(topic, msg);
        if (null != callback) {
            future.addCallback(callback);
        }
    }
}

package com.example.czl.kafka.kafka.producer.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


@Slf4j
@Component
@RequiredArgsConstructor
public class ConsumerDemo {
    @KafkaListener(topics = "test-topic-1")
    public void receivingMsg(String msg) {
        log.info("接收到Kafka消息 - msg : {}", msg);
    }
}

package com.example.czl.kafka.controller;

import com.example.czl.kafka.kafka.producer.ProducerDemo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@Slf4j
@RestController
@RequiredArgsConstructor
@RequestMapping("/test")
public class TestController {
    private final ProducerDemo producerDemo;

    @GetMapping("/send/kafka_msg")
    public Long sendMsg(String msg) {
        log.info("测试发送kafka消息 - msg : {}", msg);
        producerDemo.send("test-topic-1", msg, null);
        return System.currentTimeMillis();
    }
}