微服务中的Kafka与Micronaut
2019-11-27 16:02:02来源:博客园 阅读 ()
微服务中的Kafka与Micronaut
今天,我们将通过Apache Kafka
主题构建一些彼此异步通信的微服务。我们使用Micronaut
框架,它为与Kafka
集成提供专门的库。让我们简要介绍一下示例系统的体系结构。我们有四个微型服务:订单服务
,行程服务
,司机服务
和乘客服务
。这些应用程序的实现非常简单。它们都有内存存储,并连接到同一个Kafka
实例。
我们系统的主要目标是为客户安排行程。订单服务应用程序还充当网关。它接收来自客户的请求,保存历史记录并将事件发送到orders
主题。所有其他微服务都在监听orders
这个主题,并处理order-service
发送的订单。每个微服务都有自己的专用主题,其中发送包含更改信息的事件。此类事件由其他一些微服务接收。架构如下图所示。
在阅读本文之前,有必要熟悉一下Micronaut
框架。您可以阅读之前的一篇文章,该文章描述了通过REST API构建微服务通信的过程
:使用microaut框架构建微服务的快速指南。
1 运行Kafka
要在本地机器上运行Apache Kafka
,我们可以使用它的Docker映像。最新的镜像是由https://hub.docker.com/u/wurstmeister共享的。在启动Kafka
容器之前,我们必须启动kafka
所用使用的ZooKeeper
服务器。如果在Windows
上运行Docker
,其虚拟机的默认地址是192.168.99.100
。它还必须设置为Kafka
容器的环境。
Zookeeper
和Kafka
容器都将在同一个网络中启动。在docker中运行Zookeeper以zookeeper
的名称提供服务,并在暴露2181
端口。Kafka
容器需要在环境变量使用KAFKA_ZOOKEEPER_CONNECT
的地址。
$ docker network create kafka
$ docker run -d --name zookeeper --network kafka -p 2181:2181 wurstmeister/zookeeper
$ docker run -d --name kafka -p 9092:9092 --network kafka --env KAFKA_ADVERTISED_HOST_NAME=192.168.99.100 --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 wurstmeister/kafka
2 添加micronaut-kafka依赖
使用Kafka
构建的microaut
应用程序可以在HTTP服务器存在的情况下启动,也可以在不存在HTTP服务器的情况下启动。要启用Micronaut Kafka
,需要添加micronaut-kafka
库到依赖项。如果您想暴露HTTP API
,您还应该添加micronaut-http-server-netty
:
<dependency>
<groupId>io.micronaut.configuration</groupId>
<artifactId>micronaut-kafka</artifactId>
</dependency>
<dependency>
<groupId>io.micronaut</groupId>
<artifactId>micronaut-http-server-netty</artifactId>
</dependency>
3 构建订单微服务
订单微服务
是唯一一个启动嵌入式HTTP服务器并暴露REST API
的应用程序。这就是为什么我们可以为Kafka
提供内置Micronaut
健康检查。要做到这一点,我们首先应该添加micronaut-management
依赖:
<dependency>
<groupId>io.micronaut</groupId>
<artifactId>micronaut-management</artifactId>
</dependency>
为了方便起见,我们将通过在application.yml
中定义以下配置来启用所有管理端点并禁用它们的HTTP身份验证。
endpoints:
all:
enabled: true
sensitive: false
现在,可以在地址栏http://localhost:8080/health下使用health check
。我们的示例应用程序还将暴露添加新订单
和列出所有以前创建的订单
的简单REST API
。下面是暴露这些端点的Micronaut
控制器实现:
@Controller("orders")
public class OrderController {
@Inject
OrderInMemoryRepository repository;
@Inject
OrderClient client;
@Post
public Order add(@Body Order order) {
order = repository.add(order);
client.send(order);
return order;
}
@Get
public Set<Order> findAll() {
return repository.findAll();
}
}
每个微服务都使用内存存储库实现。以下是订单微服务(Order-Service)
中的存储库实现:
@Singleton
public class OrderInMemoryRepository {
private Set<Order> orders = new HashSet<>();
public Order add(Order order) {
order.setId((long) (orders.size() + 1));
orders.add(order);
return order;
}
public void update(Order order) {
orders.remove(order);
orders.add(order);
}
public Optional<Order> findByTripIdAndType(Long tripId, OrderType type) {
return orders.stream().filter(order -> order.getTripId().equals(tripId) && order.getType() == type).findAny();
}
public Optional<Order> findNewestByUserIdAndType(Long userId, OrderType type) {
return orders.stream().filter(order -> order.getUserId().equals(userId) && order.getType() == type)
.max(Comparator.comparing(Order::getId));
}
public Set<Order> findAll() {
return orders;
}
}
内存存储库存储Order
对象实例。Order
对象还被发送到名为orders
的Kafka主题。下面是Order
类的实现:
public class Order {
private Long id;
private LocalDateTime createdAt;
private OrderType type;
private Long userId;
private Long tripId;
private float currentLocationX;
private float currentLocationY;
private OrderStatus status;
// ... GETTERS AND SETTERS
}
4 使用Kafka异步通信
现在,让我们想一个可以通过示例系统实现的用例——添加新的行程
。
我们创建了OrderType.NEW_TRIP
类型的新订单。在此之后,(1)订单服务
创建一个订单并将其发送到orders
主题。订单由三个微服务接收:司机服务
、乘客服务
和行程服务
。
(2)所有这些应用程序都处理这个新订单。乘客服务
应用程序检查乘客帐户上是否有足够的资金。如果没有,它就取消了行程,否则它什么也做不了。司机服务
正在寻找最近可用的司机,(3)行程服务
创建和存储新的行程。司机服务
和行程服务
都将事件发送到它们的主题(drivers
, trips
),其中包含相关更改的信息。
每一个事件可以被其他microservices
访问,例如,(4)行程服务
侦听来自司机服务
的事件,以便为行程分配一个新的司机
下图说明了在添加新的行程时,我们的微服务之间的通信过程。
现在,让我们继续讨论实现细节。
4.1 发送订单
首先,我们需要创建Kafka 客户端,负责向主题发送消息。我们创建的一个接口,命名为OrderClient
,为它添加@KafkaClient
并声明用于发送消息的一个或多个方法。每个方法都应该通过@Topic
注解设置目标主题名称。对于方法参数,我们可以使用三个注解@KafkaKey
、@Body
或@Header
。@KafkaKey
用于分区,这是我们的示例应用程序所需要的。在下面可用的客户端实现中,我们只使用@Body
注解。
@KafkaClient
public interface OrderClient {
@Topic("orders")
void send(@Body Order order);
}
4.2 接收订单
一旦客户端发送了一个订单,它就会被监听orders
主题的所有其他微服务接收。下面是司机服务
中的监听器实现。监听器类OrderListener
应该添加@KafkaListener
注解。我们可以声明groupId
作为一个注解参数,以防止单个应用程序的多个实例接收相同的消息。然后,我们声明用于处理传入消息的方法。与客户端方法相同,应该通过@Topic
注解设置目标主题名称,因为我们正在监听Order
对象,所以应该使用@Body
注解——与对应的客户端方法相同。
@KafkaListener(groupId = "driver")
public class OrderListener {
private static final Logger LOGGER = LoggerFactory.getLogger(OrderListener.class);
private DriverService service;
public OrderListener(DriverService service) {
this.service = service;
}
@Topic("orders")
public void receive(@Body Order order) {
LOGGER.info("Received: {}", order);
switch (order.getType()) {
case NEW_TRIP -> service.processNewTripOrder(order);
}
}
}
4.3 发送到其他主题
现在,让我们看一下司机服务
中的processNewTripOrder
方法。DriverService
注入两个不同的Kafka Client
bean: OrderClient
和DriverClient
。当处理新订单时,它将试图寻找与发送订单的乘客最近的司机。找到他之后,将该司机的状态更改为UNAVAILABLE
,并将带有Driver
对象的事件发送到drivers
主题。
@Singleton
public class DriverService {
private static final Logger LOGGER = LoggerFactory.getLogger(DriverService.class);
private DriverClient client;
private OrderClient orderClient;
private DriverInMemoryRepository repository;
public DriverService(DriverClient client, OrderClient orderClient, DriverInMemoryRepository repository) {
this.client = client;
this.orderClient = orderClient;
this.repository = repository;
}
public void processNewTripOrder(Order order) {
LOGGER.info("Processing: {}", order);
Optional<Driver> driver = repository.findNearestDriver(order.getCurrentLocationX(), order.getCurrentLocationY());
driver.ifPresent(driverLocal -> {
driverLocal.setStatus(DriverStatus.UNAVAILABLE);
repository.updateDriver(driverLocal);
client.send(driverLocal, String.valueOf(order.getId()));
LOGGER.info("Message sent: {}", driverLocal);
});
}
// ...
}
这是Kafka Client
在司机服务
中的实现,用于向driver
主题发送消息。因为我们需要将Driver
与Order
关联起来,所以我们使用@Header
注解 的orderId
参数。没有必要把它包括到Driver
类中,将其分配给监听器端的正确行程。
@KafkaClient
public interface DriverClient {
@Topic("drivers")
void send(@Body Driver driver, @Header("Order-Id") String orderId);
}
4.4 服务间通信
由DriverListener
收到@KafkaListener
在行程服务
中声明。它监听传入到trip
主题。接收方法的参数和客户端发送方法的类似,如下所示:
@KafkaListener(groupId = "trip")
public class DriverListener {
private static final Logger LOGGER = LoggerFactory.getLogger(OrderListener.class);
private TripService service;
public DriverListener(TripService service) {
this.service = service;
}
@Topic("drivers")
public void receive(@Body Driver driver, @Header("Order-Id") String orderId) {
LOGGER.info("Received: driver->{}, header->{}", driver, orderId);
service.processNewDriver(driver);
}
}
最后一步,将orderId
查询到的行程Trip
与driverId
关联,这样整个流程就结束。
@Singleton
public class TripService {
private static final Logger LOGGER = LoggerFactory.getLogger(TripService.class);
private TripInMemoryRepository repository;
private TripClient client;
public TripService(TripInMemoryRepository repository, TripClient client) {
this.repository = repository;
this.client = client;
}
public void processNewDriver(Driver driver, String orderId) {
LOGGER.info("Processing: {}", driver);
Optional<Trip> trip = repository.findByOrderId(Long.valueOf(orderId));
trip.ifPresent(tripLocal -> {
tripLocal.setDriverId(driver.getId());
repository.update(tripLocal);
});
}
// ... OTHER METHODS
}
5 跟踪
我们可以使用Micronaut Kafka轻松地启用分布式跟踪。首先,我们需要启用和配置Micronaut跟踪。要做到这一点,首先应该添加一些依赖项:
<dependency>
<groupId>io.micronaut</groupId>
<artifactId>micronaut-tracing</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-instrumentation-http</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>io.opentracing.brave</groupId>
<artifactId>brave-opentracing</artifactId>
</dependency>
<dependency>
<groupId>io.opentracing.contrib</groupId>
<artifactId>opentracing-kafka-client</artifactId>
<version>0.0.16</version>
<scope>runtime</scope>
</dependency>
我们还需要在application.yml
配置文件中,配置Zipkin 的追踪的地址等
tracing:
zipkin:
enabled: true
http:
url: http://192.168.99.100:9411
sampler:
probability: 1
在启动应用程序之前,我们必须运行Zipkin
容器:
$ docker run -d --name zipkin -p 9411:9411 openzipkin/zipkin
6 总结
在本文中,您将了解通过Apache Kafka
使用异步通信构建微服务架构的过程。我已经向大家展示了Microaut Kafka
库最重要的特性,它允许您轻松地声明Kafka
主题的生产者和消费者,为您的微服务启用健康检查
和分布式跟踪
。我已经为我们的系统描述了一个简单的场景的实现,包括根据客户的请求添加一个新的行程。本示例系统的整体实现,请查看GitHub上的源代码
原文链接:https://www.cnblogs.com/liululee/p/11943720.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
- MyBatis中的$和#,用不好,准备走人! 2020-06-11
- 数据源管理 | Kafka集群环境搭建,消息存储机制详解 2020-06-11
- 构造函数中的this()和super() 2020-06-10
- 聊聊微服务架构及分布式事务解决方案! 2020-06-10
- java环境教程:Tomcat下载,安装,设置为Windows服务,启动 2020-06-09
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash