使用Micronaut访问Apache Kafka
微型凡人卡夫卡
由于Micronaut声称具有对Apache Kafka的支持，所以我决定试一试。
卡夫卡技术支持
据说可以使用Micronaut来创建Apache Kafka的Producer和Consumer。
微服务Kafka
环境
这次使用的Micronaut版本。
$ mn -V
| Micronaut Version: 1.0.4
| JVM Version: 1.8.0_191
在Micronaut 1.0.4版本中，似乎支持Apache Kafka 2.0.1版本。
升级到Kafka 2.0.1
微服务级别的Kafka
根据这个要求，我们下载了Apache Kafka 2.0.1。
$ wget http://ftp.tsukuba.wide.ad.jp/software/apache/kafka/2.0.1/kafka_2.12-2.0.1.tgz
$ tar xf kafka_2.12-2.0.1.tgz
$ cd kafka_2.12-2.0.1
按照Apache Kafka的快速入门，启动Apache ZooKeeper和Apache Kafka Broker。
快速启动/启动服务器
## Apache Zookeeper
$ bin/zookeeper-server-start.sh config/zookeeper.properties
## Apache Kafka（Broker）
$ bin/kafka-server-start.sh config/server.properties
这个主题是用“my-topic”作为名称创建的。
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic my-topic
Created topic "my-topic".
制作应用程序
那么，我来试着创建一个应用程序。创建了一个供应者和消费者 （这里以监听器表示）用的项目。
## Producer
$ mn create-app --features kafka --build maven hello-kafka-producer
## Listener（Consumer）
$ mn create-app --features kafka --build maven hello-kafka-listener
随后，将逐步向这两个项目添加和编辑源代码。
制造一个制作人
那么，让我们来编写一个向Apache Kafka Broker发送消息的Producer。
$ cd hello-kafka-producer
主类将保持自动生成的状态使用。
src/main/java/hello/kafka/producer/Application.java的中文翻译可以是：主要/主文件/入口文件/起始文件/应用程序.java
package hello.kafka.producer;
import io.micronaut.runtime.Micronaut;
public class Application {
public static void main(String[] args) {
Micronaut.run(Application.class);
}
}
生产者似乎需要创建带有@KafkaClient注释的接口。
定义 @KafkaClient 方法
以这种方式创建，不需要创建与接口相对应的实现类。
src/main/java/hello/kafka/producer/MessageClient.java 的原文件。
package hello.kafka.producer;
import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.reactivex.Single;
@KafkaClient(acks = KafkaClient.Acknowledge.ALL)
public interface MessageClient {
@Topic("my-topic")
Single<String> send(@KafkaKey String key, Single<String> message);
}
配置可以在@KafkaClient注释和配置文件中进行。
在@KafkaClient注解中，只进行了ACK的设置。
@KafkaClient(acks = KafkaClient.Acknowledge.ALL)
public interface MessageClient {
设置文件在这里。
src/main/resources/application.yml 的原文件路径。
---
micronaut:
application:
name: hello-kafka-producer
---
kafka:
bootstrap:
servers: localhost:9092
producers:
default:
retries: 5
在kafka.bootstrap.servers中进行连接配置，用于与Apache Kafka Broker建立连接。
Producer的行为是通过kafka.producers.[client-id]来进行的。
通过指定ID，可以对@KafkaClient单元进行设置。
根据@KafkaClient生产者属性
因为这次没有特别指定ID，所以默认使用了名为”default”的名称。
发送消息需要使用带有@Topic注解和指定主题名称的方法。
@Topic("my-topic")
Single<String> send(@KafkaKey String key, Single<String> message);
如果您要指定键，则可以使用@KafkaKey。如果不使用键，则无需指定（可以省略键参数本身）。
另外，由于可以使用RxJava等Reactive Streams相关的类型，因此我们选择了这个选项。
响应式和非阻塞方法定义
最后，使用 @KafkaClient 的 @Controller。
src/main/java/hello/kafka/producer/MessageController.java 的源代码文件位置
package hello.kafka.producer;
import javax.inject.Inject;
import io.micronaut.http.MediaType;
import io.micronaut.http.annotation.Body;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Post;
import io.reactivex.Single;
@Controller("/message")
public class MessageController {
@Inject
MessageClient messageClient;
@Post(value = "/{key}", consumes = MediaType.TEXT_PLAIN, produces = MediaType.TEXT_PLAIN)
public Single<String> message(String key, @Body Single<String> value) {
return messageClient.send(key, value).map(v -> String.format("message [%s] sended", v));
}
}
构建。
$ ./mvnw package
现在，制片方已经完成了。
听众（消费者）
接下来，让我们来谈谈消费者方面。
$ cd hello-kafka-listener
使用 @KafkaListener 的 Kafka 消费者
消费者可以使用 @KafkaListener 注解来创建。这是一个类的定义。
src/main/java/hello/kafka/listener/MessageListener.java：
主要的JAVA文件，监听Kafka消息的监听器文件。
package hello.kafka.listener;
import java.util.List;
import io.micronaut.configuration.kafka.Acknowledgement;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.OffsetReset;
import io.micronaut.configuration.kafka.annotation.OffsetStrategy;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.reactivex.Single;
@KafkaListener(groupId = "message-group", offsetReset = OffsetReset.EARLIEST, offsetStrategy = OffsetStrategy.DISABLED)
public class MessageListener {
@Topic("my-topic")
public void receive(@KafkaKey String key, Single<String> message, Acknowledgement acknowledgement) {
message.subscribe(m -> {
System.out.printf("Received key / message = %s / %s%n", key, m);
acknowledgement.ack();
});
}
}
使用@KafkaListener注解来配置消费者，并在接收消息的方法上指定@Topic注解。
@KafkaListener(groupId = "message-group", offsetReset = OffsetReset.EARLIEST, offsetStrategy = OffsetStrategy.DISABLED)
public class MessageListener {
@Topic("my-topic")
public void receive(@KafkaKey String key, Single<String> message, Acknowledgement acknowledgement) {
使用的类型是反应式的。
接收和返回反应式类型
此外，我们还使用ACK。
message.subscribe(m -> {
System.out.printf("Received key / message = %s / %s%n", key, m);
acknowledgement.ack();
});
这次，设置仅限于连接到经纪人。另外，由于在同一主机上启动了Producer，所以将micronaut.server.port设置为8080以外的值。
源代码/主要资源/应用程序配置文件/application.yml
---
micronaut:
application:
name: hello-kafka-listener
server:
port: 9080
---
kafka:
bootstrap:
servers: localhost:9092
这也是一个项目。
$ ./mvnw package
现在，消费者这方面也准备就绪了。
试着动一下。
那么，我们来进行动作确认吧。
## Producerを起動
$ java -jar target/hello-kafka-producer-0.1.jar
## Consumerを起動
$ java -jar target/hello-kafka-listener-0.1.jar
我会随便给制片人发一条消息试试看。
$ curl -H 'Content-Type: text/plain' localhost:8080/message/key1 -d 'value1'
message [value1] sended
$ curl -H 'Content-Type: text/plain' localhost:8080/message/key2 -d 'value2'
message [value2] sended
$ curl -H 'Content-Type: text/plain' localhost:8080/message/key3 -d 'value3'
message [value3] sended
在消费者端，接收到的消息将以这种方式显示。
Received key / message = key1 / value1
Received key / message = key2 / value2
Received key / message = key3 / value3
看起来你成功地移动了。
