SpringBoot集成Kafka，实现简单的收发消息

实现之前一定要把JDK,Zookeeper和Kafka都配置好

需要先配置下Kafka

在kafka的config目录下找到server.properties配置文件

image.png

把listeners和 advertised.listeners两处配置的注释去掉，可以根据需要配置连接的服务器外网IP和端口号，我这里演示选择的是本地localhost和默认端口9092

Kafka与SpringBoot进行整合

1.引入依赖

<dependency>
     <groupId>org.springframework.kafka</groupId>
     <artifactId>spring-kafka</artifactId>
</dependency>

2.编写生产者和消费者

@RestController
public class KafkaController {
    private static Logger logger = LoggerFactory.getLogger(KafkaController.class);
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/kafka")
    public String testKafka() {
        int iMax = 100;
        for (int i = 1; i < iMax; i++) {
            kafkaTemplate.send("test","key" + i, "data" + i);
        }
        return "success";
    }

    @KafkaListener(topics = "test")
    public void receive(ConsumerRecord<?, ?> consumer) {
        logger.info("{} - {}:{}", consumer.topic(), consumer.key(), consumer.value());
    }
}

相关代码说明

KafkaTemplate这个类包装了个生产者Producer，来提供方便的发送数据到kafka的主题topic里面。
send()方法的源码，KafkaTemplate类中还重载了很多send()方法，有需要可以看看源码

public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
        ProducerRecord<K, V> producerRecord = new ProducerRecord(topic, key, data);
        return this.doSend(producerRecord);
    }

通过KafkaTemplate模板类发送数据。
kafkaTemplate.send(String topic, K key, V data),第一个入参是主题，第二个入参是发送的对象，第三个入参是发送的数据。通过@KafkaListener注解配置用户监听topics

配置文件application.yml 

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: kafka2
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

bootstrap-servers：kafka服务器地址(可以多个)
consumer.group-id:指定一个默认的组名
不指定的话会报

java.lang.IllegalStateException: No group.id found in consumer config, 
container properties, or @KafkaListener annotation; 
a group.id is required when group management is used.
auto-offset-reset：自动偏移量

1.earliest ：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
2.latest ：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
3.none ：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
这个属性也是必须配置的，不然也是会报错的

org.apache.kafka.common.config.ConfigException: 
Invalid value  for configuration auto.offset.reset: 
String must be one of: latest, earliest, none

消息序列化和反序列化

在使用Kafka发送接收消息时，生产者producer端需要序列化，消费者consumer端需要反序列化，由于网络传输过来的是byte[]，只有反序列化后才能得到生产者发送的真实的消息内容。这样消息才能进行网络传输
consumer.key-deserializer和consumer.value-deserializer是消费者key/value反序列化
producer.key-deserializer和producer.value-deserializer是生产者key/value序列化

StringDeserializer是内置的字符串反序列化方式

public class StringDeserializer implements Deserializer<String> {
  public String deserialize(String topic, byte[] data) {
        try {
        //如果数据为空，那么直接返回null即可,否则将byte[]反序列化，即转为String即可
            return data == null ? null : new String(data, this.encoding);
        } catch (UnsupportedEncodingException var4) {
            throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + this.encoding);
        }
    }
   ......
}

StringSerializer是内置的字符串序列化方式

public class StringSerializer implements Serializer<String> {
    public byte[] serialize(String topic, String data) {
        try {
        //如果数据为空，那么直接返回null即可,否则将String序列化，即转为byte[]即可
            return data == null ? null : data.getBytes(this.encoding);
        } catch (UnsupportedEncodingException var4) {
            throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + this.encoding);
        }
    }
    ......
}

在org.apache.kafka.common.serialization源码包中还提供了多种类型的序列化和反序列化方式
要自定义序列化方式，需要实现接口Serializer
要自定义反序列化方式，需要实现接口Deserializer
详细可以参考
https://blog.csdn.net/shirukai/article/details/82152172

启动项目进行测试

这是Kafka的消费者Consumer的配置信息，每个消费者都会输出该配置信息

    auto.commit.interval.ms = 5000
    auto.offset.reset = earliest
    bootstrap.servers = [localhost:9092]
    check.crcs = true
    client.id = 
    connections.max.idle.ms = 540000
    default.api.timeout.ms = 60000
    enable.auto.commit = true
    exclude.internal.topics = true
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    group.id = kafka2
    heartbeat.interval.ms = 3000
    interceptor.classes = []
    internal.leave.group.on.close = true
    isolation.level = read_uncommitted
    key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    session.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

2018-11-22 14:16:53.465  INFO 11980 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 2.0.0
2018-11-22 14:16:53.465  INFO 11980 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : 3402a8361b734732
2018-11-22 14:16:57.664  INFO 11980 --- [           main] org.apache.kafka.clients.Metadata        : Cluster ID: d3n7Snc2TFmSFcNsHjqgVw

访问http://localhost:8080/kafka,就可以看到控制台打印消息了

image.png

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • Kafka史上最详细原理总结
    姓名：周小蓬 16019110037 转载自：http://blog.csdn.net/YChenFeng/art...
    aeytifiw阅读 34,768评论 13赞 425
  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 135,027评论 19赞 139
  • Kafka架构和原理深度剖析
    Kafka简介 Kafka是一种分布式的，基于发布/订阅的消息系统。主要设计目标如下： 以时间复杂度为O(1)的方...
    Alukar阅读 3,109评论 0赞 43
  • kafka 4. 设计思想（临时）
    4. 设计思想 4.1 动机 我们设计的 Kafka 能够作为一个统一的平台来处理大公司可能拥有的所有实时数据馈送...
    疯狂的橙阅读 1,096评论 1赞 4
  • 030再分配
    -->TOP1概念: 在基础收入上，政府为了社会公平，通过各种方式，实现财富转移的一个的过程。 -->TOP2问题...
    瑾兰阅读 477评论 0赞 1