使用Amplication，GraphQL，MongoDB和Kafka建筑可扩展的微服务-DEV365 开发者社区

介绍

微服务是一种架构方法，在该方法中，软件被解耦和由独立的小型服务或模块组成，并与API相互通信。每个模块都支持业务逻辑并管理自己的数据库。由于每个服务都与彼此分离，因此可以独立隔离，维护和管理。如果服务崩溃或经历停机时间，其他人将继续工作。

您可以重建，重建，重新启动和重新部署崩溃的服务而不会损害整个应用程序。

微服务应用中的故障很容易检测和隔离。微服务需要更少的构建时间，具有可重复使用的潜力，易于扩展，并且可以使用Docker等容器技术部署。此外，服务是分散，专业和自治的。

另一方面，整体应用很难隔离和故障排除。每当出现问题时，整个应用程序都会受到影响。与微服务应用相比，这使得很难扩展。

在本文中，您将学习如何使用GraphQL，MongoDB和Kafka使用Amplication来构建微服务，以生成您的应用程序并更快地构建它。

放大是更好，更快地构建质量node.js应用程序的工具。

更快，更好地构建微服务

存在生成微服务的工具，以加速和简化基于微服务的应用程序的开发，避免重复任务，减少样板代码以及通过在应用程序中促进代码一致性和标准化来实施最佳实践。

从头开始构建微服务架构需要额外的开发工作，因为开发人员必须手动处理样板代码，配置，配置数据库模型和连接以及设置身份验证等任务。启动。

存在像Amplication这样的工具，可以帮助开发人员更快，更好地构建微服务，而无需牺牲性能，安全性或可扩展性。

扩增可以通过为您提供以下功能来帮助您构建微服务。

项目脚手架 - 放大使您可以扩展现有项目或脚手架新的微服务。它根据您的需求生成用于微服务，文件夹结构和配置文件的应用程序结构和样板代码。
代码生成 - 通过自动为微服务生成样板和基础架构代码来减少手动编码工作，并确保在微服务之间保持一致性。生成的功能包括REST API端点，数据库访问，身份验证，基于角色的许可，实体等。这些是根据您在Amplication Dashboard上的架构配置生成的。

每当您在仪表板中进行更改时，就会生成应用程序的新版本。

。

服务通信 - 放大允许您设置服务间通信。默认情况下，Kafka插件包含在放大中。它可以安装在您的项目中，并用于在服务之间进行通信。
部署自动化 - 随着放大，您可以通过插件连接到CI/CD工具和云提供商或生成Docker容器，从而使始终有效地部署微服务更容易。
>

要求

要遵循本教程，您需要：

An Amplication account
对node.js，graphql，mongodb和kafka的基本理解
Docker安装在您的机器上。

通过放大设置项目

按照以下步骤来创建一个通过放大的项目。

登录Amplication或创建一个新帐户，如果您还没有一个。
创建服务。
选择要如何构建服务。
我们在本文中使用GraphQl API。

选择一个回购类型。我们将用于本教程的目的。

选择MongoDB作为选择的数据库

在您的项目中添加身份验证

之后，放大将生成您的服务。

您的项目已设置 - 您现在可以为应用程序创建实体。

创建数据库实体

让S创建实体设置我们的数据库。对于本教程，除默认用户实体外，我们还将仅创建一个称为帖子的实体。

导航到仪表板上的“实体”选项卡。选择Add entity。

单击“添加”字段添加字段。我们将创建三个字段值，例如* ，*和uid。 Value的数据类型是多行文本，Like是boolean，而UID与User entity有关，因此我们可以获取属于特定用户的帖子。

创建角色

放大使您可以创建角色并授予它们权限。这对于通过根据用户的角色授予权限或访问权限来确保应用程序安全很重要。例如，某些特权（例如删除或阅读字段或从组中删除用户）可能保留给管理员，并仅限于常规用户。允许CRUD访问每个实体可以构成安全威胁。

要创建角色，请单击Roles图标，然后选择“添加角色”。给它任何您想要的名字 - 在这里，我们称其为Poster。

切换User实体中每个命令的权限，以便只有User帐户将具有User角色。默认情况下，放大将每个角色设置为对每个实体都有CRUD访问。

处理身份验证

创建一个用户

我们希望使新用户能够创建一个帐户。回想一下，我们修改了权限，以便只有具有User角色的用户才能访问User实体来创建，读取，更新或删除条目。

让我们编写逻辑以创建新用户。

server/src/auth/auth.service.ts

async signup(credentials: Credentials): Promise<UserInfo>{
    const {username, password } = credentials;


    const user = await this.userService.create({
      data: {
        username,
        password,
        roles: ["poster"]
      }
    })

    if(!user) {
      throw new UnauthorizedException("Error creating user")
    }
    const accessToken = await this.tokenService.createToken({
      id: user.id,
      username,
      password
    })
    return {
      accessToken,
      username: user.username,
      id: user.id,
      roles: (user.roles as {roles: string[]}).roles
    }
  }

如果注册成功，它将返回accessToken，username，id和roles。

我们需要使用 @Mutation装饰器在GraphQl Server中将注册设置为突变。进入server/src/auth/auth.resolver.ts并添加以下代码行：

@Mutation(() => UserInfo)
  async signup(@Args() args: LoginArgs): Promise<UserInfo> {
    return this.authService.signup(args.credentials);
  }

获取签名用户

我们希望能够获取签名用户的信息。让我们通过用户名在用户实体中找到用户。为此，请将以下代码行添加到tokenservice类。

server/src/auth/token.service.ts

decodeToken(bearer: string): string {
        return this.jwtService.verify(bearer).username;
    }

让我们编写逻辑以在Authservice类中获取签名用户。

server/src/auth/auth.service.ts

import { User } from "../user/base/User"


 async me(authorization: string = ""): Promise<User> {
    const bearer = authorization.replace(/^Bearer\s/, "");
    const username = this.tokenService.decodeToken(bearer);
    const result = await this.userService.findOne({
      where: { username },
      select: {
        createdAt: true,
        firstName: true,
        id: true,
        lastName: true,
        roles: true,
        updatedAt: true,
        username: true,
      },
    })


    if(!result){
      throw new NotFoundException(`We couldn't anything for ${username}`)
    }
    return result
  }

me方法解码JWT令牌以获取用户名，然后使用用户名查找属于已登录的用户的其他信息。

让我们像使用signup方法一样，将新的me方法添加到AuthResolver类中。

server/src/auth/auth.resolver.ts

import * as common from "@nestjs/common";
 import { Args, Mutation, Query, Resolver, Context } from "@nestjs/graphql";
 import { Request } from "express"
 import * as gqlACGuard from "../auth/gqlAC.guard";
 import { AuthService } from "./auth.service";
 import { GqlDefaultAuthGuard } from "./gqlDefaultAuth.guard";
 import { UserData } from "./userData.decorator";
 import { LoginArgs } from "./LoginArgs";
 import { UserInfo } from "./UserInfo";
 import { User } from "../user/base/User"


export class AuthResolver {
  constructor(private readonly authService: AuthService) {}
    @Query(() => User)
    async me(@Context('req') request: Request): Promise<User> {
            return this.authService.me(request.headers.authorization);
    }
 }

用GraphQl提出CRUD请求

通过GraphQl注册

通过http://localhost:3000/graphql导航到GraphQl操场，让我们注册一个新用户。

mutation signup($credentials: Credentials!) {
    signup(credentials: $credentials) {
      accessToken
        username
    }
  }

将帐户凭据（用户名和密码）添加到Query Variables字段。

 {"credentials": {
    "username": "janedoe",
  "password": "000000"
}}

成功注册后，我们将收到accessToken和username，如我们的查询中指定。我们还可以查询其他字段，例如id和roles。

通过GraphQl登录

如果登录凭据正确，我们希望将accessToken用于提出未来的授权请求。

mutation login($credentials: Credentials!) {
    login(credentials: $credentials) {
      accessToken
    }
  }

将凭据添加到Query Variables字段。

{"credentials": {
    "username": "johndoe",
      "password": "123456"
  }}

查询签名用户

让您查询当前用户并获取其username和id。

query me{
  me {
    id
    username

  }
}

将授权令牌添加到GraphQl Playground的HTTP标头字段中。

{"Authorization": "Bearer add-your-token-here"}

创建新帖子

让我们通过GraphQl创建新帖子。

mutation createPost ($data: PostCreateInput!){
  createPost(data: $data){
    value
    like
    id
  }
  
}

查询变量必须包含value，因为它是必需的字段，并且不是自动生成的。

{ "data": {"like": true, "value": "my first post"}}

记住添加您的令牌。

获取所有帖子

让我们查询数据库中的所有帖子（无论如何我们现在只有一个）。

query posts($where: PostWhereInput, $orderBy: [PostOrderByInput!]) {
    posts(where: $where, orderBy: $orderBy) {
      value
    like
    id
    }
  }

设置Kafka以在服务之间进行交流

kafka是一个分布式事件流平台，旨在实时处理数据提要。

它提供了高度可扩展，容忍和耐用的出版物订阅消息传递系统。 KAFKA广泛用于现代数据体系结构，用作构建实时数据管道和流媒体应用程序的骨干。 KAFKA用于日志聚合，事件采购，流处理，消息传递系统和提交日志。

关键概念：

主题：消息在卡夫卡中组织为主题。每个主题都是可以写入或消耗的记录流。

生产者：生产者的任务是向Kafka主题发布消息。
消费者：消费者的工作是阅读主题中存在的记录
经纪人：经纪人是负责存储和复制消息的服务器集群。
分区：主题分为一个或多个分区。

通过放大安装Kafka插件

放大提供了一些插件，您可以使用这些插件来扩展应用程序的功能。这些插件之一是Kafka。让我们从放大仪表板上安装kafka插件，然后开始在我们的应用程序中使用kafka。

进入仪表板上的插件页面。单击所有插件，然后继续安装Kafka插件。将更改提交给github。

我们还需要创建一个新的消息经纪。

转到您的放大仪表板，单击Add Resource，然后从列表中选择Message Broker。
添加主题

单击连接在服务仪表板上，启用消息经纪。

设置Kafka生产商

让我们在我们的应用中设置一个Kafka生产商。制作人负责在Kafka主题上撰写消息。

server/src/kafka/kafka.service.ts

import { Injectable, OnModuleInit, OnApplicationShutdown } from "@nestjs/common";
 import { KafkaServiceBase } from "./base/kafka.service.base";
 import { Kafka, Producer, ProducerRecord } from 'kafkajs';


  @Injectable()
  export class KafkaService extends KafkaServiceBase implements OnModuleInit, 
  OnApplicationShutdown {

    private readonly kafka = new Kafka({
        brokers: ['localhost:9092']
    });

    private readonly producer: Producer = this.kafka.producer()

    //connect producer to our server
    async onModuleInit() {
        await this.producer.connect()
    }


    //produce messages
    async produce(record: ProducerRecord){
        await this.producer.send(record)

     }


     //disconnect producer on application shutdown
     async onApplicationShutdown(){
        await this.producer.disconnect();
     }
}

onModuleInit，我们的生产商连接，然后断开onApplicationShutdown。不要忘记添加KafkaService作为KafkaModule中的提供商。

创建Kafka消费者服务

让我们设置消费者服务。

server/src/kafka/consumer.service.ts

import {Injectable, OnApplicationShutdown} from "@nestjs/common";
 import { Kafka, ConsumerRunConfig, ConsumerSubscribeTopics, Consumer } from  "kafkajs"


 @Injectable()
    export class ConsumerService implements OnApplicationShutdown{
        private readonly kafka = new Kafka({
            brokers: ['localhost:9092']
        });
        private readonly consumers: Consumer[] = [];


        async consume(topic: ConsumerSubscribeTopics, config: ConsumerRunConfig){
            const consumer = this.kafka.consumer({groupId: "YOUR_KAFKA_GROUP_ID"});
            await consumer.connect();
            await consumer.subscribe(topic);
            await consumer.run(config)
            this.consumers.push(consumer)
        }


        async onApplicationShutdown() {
            for (const consumer of this.consumers){
                await consumer.disconnect()
            }
        }
    }

创建Kafka消费者

消费者将在主题中读取数据。

server/src/kafka/kafka.consumer.ts

import {  Injectable, OnModuleInit } from "@nestjs/common";
import { ConsumerService } from "./consumer.service"


@Injectable()
export class KafkaConsumer implements OnModuleInit {
    constructor( private readonly consumerService: ConsumerService){}
    async onModuleInit() {
        await this.consumerService.consume({topics: ["Sender"]}, {
            eachMessage: async ({topic, partition, message}) => {
                console.log({
                    value: message.value?.toString(),
                    topic: topic.toString(),
                    partition: partition.toString()
                })
            }
        })
    }
}

在这里，我们正在记录value，topic和partition从主题到达控制台。

让我们通过登录时广播消息来测试我们的kafka设置。

server/src/auth/auth.service.ts

export class AuthService {

  constructor(

    private readonly producerService: KafkaService

  ) {}

 await this.producerService.produce({

      topic: "Sender",

      messages: [

        {

          value: "User logged in"

        },

      ]

    })

 }

运行您的应用程序

docker-compose up
npm run start

登录时，您应该在控制台中获得value，topic和partition。

结论

放大使您可以更快，更好地构建安全，可靠，可扩展和有效的微服务。通过放大生成的应用是可扩展的。您可以扩展应用程序的功能并添加更多功能和业务逻辑。

在文章中，您学习了如何使用GraphQL，MongoDB和Kafka进行放大的微服务。您可以扩展此应用程序的功能，以使Kafka在微服务之间进行通信。

就是这个博客。希望您今天学到了一些新东西。

如果这样做，请在Twitter，LinkedIn或您的朋友或同事上喜欢/分享此内容，以便它可以吸引更多的人。

如果您是普通的读者，谢谢，您是我能够与您分享我的生活/职业经历的重要部分。

如果您发现此博客文章有帮助或有趣，请考虑将Amplication Project在GitHub上成为明星。

无耻的插头

订阅我的newsletter，以获取有关我的小赌注和写作旅程的每月更新！加入超过3500多个其他人！