1 Kafka

Kafka是一个开源分布式的流处理平台，一种高吞吐量的分布式发布订阅消息系统，它能够处理消费者在网站中的全部动做流数据。Kafka由Scala和Java编写，2012年成为Apache基金会下顶级项目。前端

2 Kafka优势

低延迟：Kafka支持低延迟消息传递，速度极快，能达到200w写/秒
高性能：Kafka对于消息的分布，订阅都有高吞吐量。即便存储了TB级的信息，依然可以保证稳定的性能
可靠性：Kafka是分布式，分区，复制和容错的，保证零停机和零数据丢失
可扩展：用户能够从但个代理Broker开始做POC，而后慢慢扩展到由三个Broker组成的小型开发集群，接着扩展到数十个甚至数百个Broker集群进入生产阶段，能够在集群联机时进行扩展，而不会影响整个系统的可用性
多个生产者：不管这些客户使用相同Topic仍是多个Topic，Kafka都能无缝处理多个生产者，使得系统能够很是容易聚合来自许多前端系统的数据并使其保持一致
多个消费者：Kafka具备多个消费者设计，能够读取任何但个消息流而不会相互干扰。多个Kafka消费者能够组成一个消费组进行操做并共享消息流，从而确保每一条消息只被整个消费组处理一次
基于磁盘的保留：Kafka使用分布式提交日志，消息可以快速持久化到磁盘上。消息持久化意味着若是消费者落后，不管是因为处理速度缓慢仍是忽然的消息涌入都不会有丢失数据的危险，也意味着消费者能够被中止。消息将保留在Kafka中，容许消费者从新启动而且从中断处获取处理信息而不会丢失数据

3 Kafka相关术语

Broker：Kafka集群包含一个或多个服务器，这种服务器称为Broker
Topic：每条发布到Kafka的消息都有一个类别，这个类别称为Topic。物理上不一样Topic的消息分开存储，逻辑上Topic的消息虽然保存在一个或多个Broker上，但用户只需指定消息的Topic便可生产或消费数据而没必要关心数据存放于何处
Partition：每一个Topic包含一个或多个Partition
Producer：生产者，负责发布消息到Broker
Consumer：消费者，向Broker读取消息的客户端
Consumer Group：每一个Consumer属于一个特定的Consumer Group，能够为每一个Consumer指定Group Name，不然属于默认Group

4 动手干活

4.1 环境

Spring Boot 2.3.1
IDEA 2020.1.1
OpenJDK 11.0.7
Kafka 2.5.0
Kotlin 1.3.72

4.2 下载Kafka

官网戳这里。java

下载并解压（注意须要Kafka与Spring Boot版本对应，能够参考这里）：git

tar -xvf kafka_2.12-2.5.0.tgz
cd kafka_2.12-2.5.0

接着启动ZooKeeper与Kafka：github

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

Kafka须要用到ZooKeeper，须要在启动Kafka以前启动ZooKeeper（ZooKeeper是一个开源的分布式应用程序协调服务，是Hadoop的组件，主要用于解决分布式应用中的一些数据管理问题）。spring

Kafka默认使用9092端口，部署在服务器上须要注意防火墙以及安全组的处理。apache

4.3 新建工程

考虑到Spring Boot在2.3.0M1中（截至本文写做日期2020.07.14Spring Boot已更新到2.4.0M1）首次采用Gradle而不是Maven来构建项目，换句话说往后Spring Boot的构建工具将从Maven迁移到Gradle，Spring Boot团队给出的主要缘由是能够减小项目构建所花费的时间，详情能够戳这里瞧瞧。bootstrap

另外因为另外一个基于JVM的语言Kotlin的日渐崛起，后端开始逐渐有人采用Kotlin（尽管很少，不过语法糖真的香，JetBrains家的语言配合IDE，爽得飞起），所以本示例项目将采用两种方式搭建：后端

Java+Maven
Kotlin+Gradle

选择的依赖以下（固然您喜欢的话能够在pom.xml或者build.gradle.kts里面加，对于Kotlin不须要Lombok）：安全

4.4 项目结构

Java版：springboot

Kotlin版：

serialize：序列化/反序列化实体类
Constant.java/Constant.kt：常量类
Consumer.java/Consumer.kt：消费者类
Entity.java/Entity.kt：实体类
Producer.java/Product.kt：生产者类
TestApplicationTests：测试类

4.5 常量类

包含Topic与GroupId，Java版：

public class Constants {
    public static final String TOPIC = "TestTopic";
    public static final String GROUP_ID = "TestGroupId";
}

Kotlin版：

object Constants
{
    const val TOPIC = "TestTopic"
    const val GROUP_ID = "TestGroupId"
}

4.6 实体类

@AllArgsConstructor
@NoArgsConstructor
@Data
@Builder
public class Entity {
    private long id;
    private String name;
    private int num;
}

说一下Lombok的几个注解：

@AllArgsConstructor/@NoArgsConstructor：生成全部参数/无参数构造方法
@Data：等价于@Setter+@Getter+@RequiredArgsConstrucotr+@ToString+@EqualAndHashCode，自动生成Setter+Getter+toString()+equals()+hashCode()，还有@RequireArgsConstructor为类的每个final或非空字段生成一个构造方法
@Builder：能够经过建造者模式建立对象

Kotlin版：

class Entity {
    var id: Long = 0
    var name: String = ""
    var num: Int = 0

    constructor()

    constructor(id:Long,name:String,num:Int)
    {
        this.id = id
        this.name = name
        this.num = num
    }
}

4.7 生产者

@Component
@Slf4j
//防止出现Field injection not recommended警告，代替了原来的直接在字段上@Autowired
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class Producer {

    private final KafkaTemplate<String, Entity> kafkaTemplate;

    public void send(Entity entity) {
    	//发送消息
    	//类型通常为String+自定义消息内容，String表明消息Topic，这里消息内容用Entity表示
        ListenableFuture<SendResult<String, Entity>> future =
                kafkaTemplate.send(Constants.TOPIC, entity);
        //回调函数
        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.info("Send message failed");
            }

            @Override
            public void onSuccess(SendResult<String, Entity> stringEntitySendResult) {
                log.info("Send message success");
            }
        });
    }
}

这里的send有两个参数，对应于sendResult<>中的参数类型，第一个为消息的Topic，第二个为消息体，通常使用String或者Json。

Kotlin版：

@Component
class Producer
{
    @Autowired
    private var kafkaTemplate:KafkaTemplate<String,Entity> ? = null
    private val log = LoggerFactory.getLogger(this.javaClass)

    fun send(entity: Entity)
    {
        val future = kafkaTemplate!!.send(Constants.TOPIC,entity);
        future.addCallback(object : ListenableFutureCallback<SendResult<String?, Entity?>?>{
            override fun onSuccess(result : SendResult<String?,Entity?>?)
            {
                log.info("Send success");
            }

            override fun onFailure(e:Throwable)
            {
                log.info("Send failed");
            }
        })
    }
}

4.8 消费者

@Component
@Slf4j
public class Consumer {
    @KafkaListener(topics = Constants.TOPIC,groupId = Constants.GROUP_ID)
    public void consume(Entity entity)
    {
        log.info("Consume a entity, id is "+entity.getId());
    }
}

使用@KafkaListener注解，第一个参数表示须要消费的消息的Topic，能够是String []，第二个是消费者组的id。生产者的消息Topic必须与消费者的Topic保持一致不然不能消费，这里简单处理打印日志。

Kotlin版：

@Component
class Consumer {
    private val log = LoggerFactory.getLogger(this.javaClass)

    @KafkaListener(topics = [Constants.TOPIC],groupId = Constants.GROUP_ID)
    fun consume(entity: Entity) {
        log.info("Consume a entity, id is "+entity.id.toString())
    }
}

4.9 序列化/反序列化

这里自定义了序列化/反序列化类，序列化/反序列化类须要实现org.apache.kafka.common.serialization.Serializer<T>/Deserializer<T>接口，其中T是想要序列化的类型，这里是Entity。序列化接口反编译以下：

public interface Serializer<T> extends Closeable {
    default void configure(Map<String, ?> configs, boolean isKey) {
    }

    byte[] serialize(String var1, T var2);

    default byte[] serialize(String topic, Headers headers, T data) {
        return this.serialize(topic, data);
    }

    default void close() {
    }
}

反序列化反编译接口以下：

public interface Deserializer<T> extends Closeable {
    default void configure(Map<String, ?> configs, boolean isKey) {
    }

    T deserialize(String var1, byte[] var2);

    default T deserialize(String topic, Headers headers, byte[] data) {
        return this.deserialize(topic, data);
    }

    default void close() {
    }
}

也就是只须要实现其中的serialize/deserialize方法便可。这里序列化/反序列化用到了自带的Jackson：

@Slf4j
public class Serializer implements org.apache.kafka.common.serialization.Serializer<Entity> {
    public byte [] serialize(String topic, Entity entity)
    {
        try {
            return entity == null ? null : new ObjectMapper().writeValueAsBytes(entity);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            log.error("Can not serialize entity in Serializer");
        }
        return null;
    }
}

反序列化：

@Slf4j
public class Deserializer implements org.apache.kafka.common.serialization.Deserializer<Entity> {
    public Entity deserialize(String topic,byte [] data)
    {
        try {
            return data == null ? null : new ObjectMapper().readValue(data,Entity.class);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("Can not deserialize entity in Deserializer");
        }
        return null;
    }
}

Kotlin版：

class Serializer : org.apache.kafka.common.serialization.Serializer<Entity?>
{
    private val log = LoggerFactory.getLogger(this.javaClass)

    override fun serialize(topic: String?, data: Entity?): ByteArray? {
        try {
            return if (data == null) null else ObjectMapper().writeValueAsBytes(data)
        }
        catch (e:JsonProcessingException)
        {
            e.printStackTrace()
            log.error("Can not serialize entity in Serializer")
        }
        return null
    }
}

class Deserializer : org.apache.kafka.common.serialization.Deserializer<Entity?>
{
    private val log = LoggerFactory.getLogger(this.javaClass)

    override fun deserialize(topic: String?, data: ByteArray?): Entity? {
        try
        {
            return ObjectMapper().readValue(data, Entity::class.java)
        }
        catch (e:IOException)
        {
            e.printStackTrace()
            log.error("Can not deserialize entity in Deserializer")
        }
        return null
    }
}

4.10 配置文件

application.properties：

# 地址，本地直接localhost，部署可使用公网ip
spring.kafka.bootstrap-servers=localhost:9092
# 消费者组id
spring.kafka.consumer.group-id=TestGroupId
spring.kafka.consumer.auto-offset-reset=earliest
# 消费者键反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费者值反序列化类
spring.kafka.consumer.value-deserializer=com.test.serialize.Deserializer

# 生产者键序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
# 生产者值序列化类
spring.kafka.producer.value-serializer=com.test.serialize.Serializer

对于auto-offest-rest，该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的状况下怎么处理，有四个取值：

earliest：当各分区有已提交的offest时，从提交的offest开始消费，无提交的offest时，从头开始消费
latest（默认）：当各分区有已提交的offest时，从提交的offest开始消费，无提交的offest时，消费新产生的该分区下的数据
none：各分区都存在已提交的offest时，从offest后消费，只要有一个分区不存在已提交的offest，则抛出异常
exception：其余状况将抛出异常给消费者

对于序列化/反序列化，String可使用自带的序列化/反序列化类：

org.apache.kafka.common.serialization.StringSerializer
org.apache.kafka.common.serialization.StringDeserializer

至于Json可使用：

org.springframework.kafka.support.serializer.JsonSerializer
org.springframework.kafka.support.serializer.JsonDeserializer

其余自定义的请实现org.apache.kafka.common.serialization.Serializer<T>/Deserializer<T>接口。

yml版：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: TestGroupId
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: com.test.serialize.Deserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: com.test.serialize.Serializer

5 测试

5.1 测试类

@SpringBootTest
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
class TestApplicationTests {

    private final Producer producer;

    @Test
    void contextLoads() {
        Random random = new Random();
        for (int i = 0; i < 1000; i++) {
            long id = i+1;
            String name = UUID.randomUUID().toString();
            int num = random.nextInt();
            producer.send(Entity.builder().id(id).name(name).num(num).build());
        }
    }
}

生产者发送1000条消息。 Kotlin版：

@SpringBootTest
class TestApplicationTests {

    @Autowired
    private val producer:Producer? = null

    @Test
    fun contextLoads() {
        for(i in 0..1000)
        {
            val id = (i + 1).toLong()
            val name = java.util.UUID.randomUUID().toString()
            val num = (0..100000).random()
            producer!!.send(Entity(id,name,num))
        }
    }
}

5.2 测试

控制台输出以下：

全部消息被成功发送而且被成功消费。

最后能够去验证一下Kafka的Topic列表，能够看到配置文件中的Topic的值（TestTopic），进入Kafka目录：

bin/kafka-topics.sh --list --zookepper localhost:2181

6 源码

7 参考

一、CSDN-Kafka优势二、简书-Spring Boot 2.x 快速集成整合消息中间件 Kafka 三、简书-springboot 之集成kafka

若是以为文章好看，欢迎点赞。

同时欢迎关注微信公众号：氷泠之路。

Spring Boot 2.x 快速集成Kafka

1 Kafka

2 Kafka优势

3 Kafka相关术语

4 动手干活

4.1 环境

4.2 下载Kafka

4.3 新建工程

4.4 项目结构

4.5 常量类

4.6 实体类

4.7 生产者

4.8 消费者

4.9 序列化/反序列化

4.10 配置文件

5 测试

5.1 测试类

5.2 测试

6 源码

7 参考