Springboot集成kafka

2022-01-25 10:32:05 阅读：125 来源： 互联网

标签：集成 Springboot ip springframework kafka 默认 import org

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.消费者

标签：集成,Springboot,ip,springframework,kafka,默认,import,org
来源： https://blog.csdn.net/hauchun/article/details/122680034

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。