1.引入依赖

        <!-- 这里kafka版本使用的是2.6.0 -->
        <kafka.version>2.6.0</kafka.version>


        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>${kafka.version}</version>
        </dependency>

2.创建配置

配置application.yaml文件

spring:
  kafka:
    # kafka 服务地址
    bootstrap-servers: ip:9092
    # 消费者配置
    consumer:
      auto-commit-interval: 5000 #自动提交消费位移时间隔时间
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 500 #批量消费每次最多消费多少条消息
      enable-auto-commit: true #开启自动提交消费位移
      auto-offset-reset: latest #其他earliest、none
      group-id: kafka.consumer.group #消费者组名称
      client-id: kafka.consumer.client.id #消费者客户端ID
      fetch-max-wait: 400 #最大等待时间
      fetch-min-size: 1 #最小消费字节数
      heartbeat-interval: 3000 #分组管理时心跳到消费者协调器之间的预计时间
      isolation-level: read_committed
      topic-name: huachun
    # 生产者配置
    producer:
      batch-size: 16384 #批次大小，默认16k
      acks: -1 #ACK应答级别，指定分区中必须要有多少个副本收到消息之后才会认为消息成功写入，默认为1只要分区的leader副本成功写入消息；0表示不需要等待任何服务端响应；-1或all需要等待ISR中所有副本都成功写入消息
      retries: 3 #重试次数
      value-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      buffer-memory: 33554432 #缓冲区大小，默认32M
      client-id: kafka.producer.client.id #客户端ID
      compression-type: none #消息压缩方式，默认为none，另外有gzip、snappy、lz4
      properties:
        retry.backoff.ms: 100 #重试时间间隔，默认100
        linger.ms: 0 #默认为0，表示批量发送消息之前等待更多消息加入batch的时间
        max.request.size: 1048576 #默认1MB，表示发送消息最大值
        connections.max.idle.ms: 540000 #默认9分钟，表示多久后关闭限制的连接
        receive.buffer.bytes: 32768 #默认32KB，表示socket接收消息缓冲区的大小，为-1时使用操作系统默认值
        send.buffer.bytes: 131072 #默认128KB，表示socket发送消息缓冲区大小，为-1时使用操作系统默认值
        request.timeout.ms: 30000 #默认30000ms，表示等待请求响应的最长时间
      topic-name: huachun

3.生产者发送消息测试

package com.hhmt.delivery;

import com.hhmt.delivery.service.ProducerService;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

/**
 * 辉煌明天
 * FileName: TestKafka
 * Author:   huachun
 * email: huachun_w@163.com
 * Date:     2022/1/24 18:45
 * Description:
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class TestKafka {

    @Autowired
    private ProducerService producerService;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Value("${spring.kafka.producer.topic-name}")
    private String topicName;

    @Test
    public void cotextLoads() {
        producerService.sendMessage(topicName, "springboot");
    }

    @Test
    public void sendMessage1() {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topicName, 0, System.currentTimeMillis(), "topic-key", "测试");
        producerRecord.headers().add("user", "zhangsan".getBytes());
        producerService.sendMessage(producerRecord);
    }

    @Test
    public void sendMessage2() {
        String event = "测试";
        Map<String, Object> map = new HashMap<>();
        map.put("user", "zhangsan");
        MessageHeaders headers = new MessageHeaders(map);
        Message<String> message = MessageBuilder.createMessage(event, headers);
        kafkaTemplate.setDefaultTopic(topicName);
        producerService.sendMessage(message);
    }

}

1.出现的问题

        此时发送kafka消息会出现一个问题

         定位后发现使用的是云主机，有内网IP和外网IP，虚拟机对外ip[暴露的ip]和真实ip[ifconfig显示的ip]可能只是映射关系，用户访问对外ip时，OpenStack会转发到对应的真实ip实现访问。但此时如果 Kafka server.properties配置中的listeners=PLAINTEXT://对外IP:9092中的ip配置为[对外ip]的时候无法启动，因为socket无法绑定监听

2.解决方案

        在kafka的server.properties中添加如下内容

listeners=PLAINTEXT://内网:9092
advertised.host.name=外网ip或者域名
advertised.listeners=PLAINTEXT://外网ip或者域名:9092

 重启kafka服务测试，消息发送成功...

参考原文：关于kafka的Cannot assign requested address_APTX4869_YXW的博客-CSDN博客

参考原文：kafka无法启动,Cannot assign requested address._MysticalYcc的博客-CSDN博客

4.消费者