?? ? ? ?? ?? Kafka? ?? ???? ??? ??? ??? ???? ???? ?????? ??????. ??? ? ??? ??? ?? ???? ??? ????. NestJS ? KafkaJS.

???? ?????

Apache Kafka? ??? ???? ????? ??? ?? ??? ???? ??????. ???? ?? ???, ?? ?? ??? ??? ???? ??, ?? ? ??? ? ???? ??? ??? ????? ? ??? ?? ??????? ???? ? ?????.

?? ??:

• ??? ????: Kafka? ??? ??? ???? ??? ???? ?????.
• ?? ????: Kafka? ???? ????? ?? ???????. ????? ??? ??? ????? ???? ?? ??? ???? ??? ? ????.
• ??-?? ??: ???? ??? ???? ?? ???? ???? ????. Kafka? ?? ???? ????? ??? ??????? ??? ??? ???? ????? ??? ? ????.
• ???: Kafka? ?? ???? ????? ?? ?? ???? ?? ??? ?? ???? ?????.
• ??? ?? ????: Kafka? ?? ??? ?? ???? ???? ???? ???? ??? ???? ???? ?????.
• ???? ? ??: ???? ?? ??? ????? ??? ????? ?? ??? ??? ?????.
• ?? ???: ???? ???? ????? ???? ?? ?? ? ??? ??? ??? ?? ??? ?????.
• ?? ? ???: Kafka? ??? ???? ???? Kafka Connect(??? ???) ? Kafka Streams(??? ???)? ?? ??? ??? ????.

??

• ???: ??? ??, ??, ????? ?? ????? ?????.
• ???: Kafka? ??? ??? ??? ???? ???? ?? ???? ??? ? ????.
• ???: ???? ?? ???? ???? ??? ???? ?????.
• ??: Kafka? ??? ??? ????? ?? ??? ???? ?????? ??? ?? ?? ??? ???? ?????.

??

??? ???? Kafka? ??? ????? ?? ???? ??? ????? ? ? ???? ??? ?? ???? ??? ? ? ????.

• ??? ???: Kafka?? ???? ?? ??? ???? ?? ?? ??? ?? ??? ?????. Kafka? ???? ??? ???? ????? ????? ???? ??? ??? ???? ???.
• ?? ????? ???? ??: Kafka? ???? ????? ????? ?? ???? ??? ???? ?? ???? ????. ??? ???? ?? ??? ?????? ?? ?? ??? ?? ? ? ????.
• ????? ?? ?? ??: Kafka? ????? ??(?: log-2024-*)? ???? ?? ???? ???? ?? ???? ????.

?? ??

• ??? ??: ??? ???? ???? ?? ???? ?????.
• ??? ??: ?????? ??? ?? ?? ??? ??? ???? ?????.
• ?? ??: ?? ????? ??? ???? ?????.
• ??? ?????: ??? ? ???? ????? ????? ???????.
• IoT ??????: IoT ??? ?? ???? ????? ?????.

???? ??? ??????

Kafka? ??? ? ??-?? ??? ??? ??? ?? ???? ????? ? ?? ??? ??? ?????.

• ???? ???? ??? ?? ??? ????? ???? ?? ??? ??? ??? ????? ?? ???? ?? ???? ???? ????.
• ??-?? ??? ?? ???? ????? ? ???? ?? ????? ????? ?? ??? ???? ?? ??? ??? ?? ????.

Kafka? ??? ?? ???? ???? ?? ? ??-?? ??? ??? ?????. ??? ??? ???? ??? ????? ?????, ? ???? ?? ?? ???(???)?? ?????. ? ??? ???? ???? ????? ?? ???? ??? ??? ? ????.

???, ?? ? ???

Kafka? ??? ???? ????? ??? ?????? ?? ???????. ??? ???? ???? ??? ?? ????? ?? ?? ?? ??? ?????.

???? ??, ???? ??, ?? ?? ? ??????? ??? ??, ?? ?? ?? ?????.

Kafka?

???? ?/? ??? ?????, ??? ?? ?? ?? ??? ???? ??????.

• ?? ???? ??? ??? ?? ??(?: ?? ?? ?? ?? ?????? ???)? ???? ??? ????. Kafka ????? ???? ?? ??? ??????.
• ?? ??? ??? ??? ? ??? ????? ??? ?? ??? ??? ??? ????. ??? ??????? ?? ??? ?? ???? ???? ???? ?? ?? ????? ?? ???, ?? ?? ??? ??? ?? ??? ?? ???? ?????.

Kafka? ???? ??? ??? ??? ???? ?????. ?? ???? Kafka? ???? ???? ?? ???? ?? ?? ?????. ???? ?? ??? ?? ??? ?? ?? ??? ?? ????. ???? ?? ??? ???? ?? ???? ????? ?? ??? ???? ????? ?????. ?, ??? ?????? ??? ? ?? ??? ?? ????.

??? ???? ?? ??? ????? ????? ??? ??? I/O ?? ??? ????. ?? ???? ?? Kafka??? ??? ????? ??? ?? ??? ?? ??? ??? ? ?? ?? ??? ?? ? ????. ??? ????? ?? Kafka? ????? ??? ? ?? ??? ???? ?? ???? ??? ? ?? ??? ?????.

Kafka? ?? ??? ??? ?? ???? ???? ?????.

• ? ??: ???? ?? ???? ??? ?? ???? ????? ???? ???? ????? ??? ??? ???? ????.
• ? ??: ???? ?? ???? ????? ??? ?? ?? ???? ?? ??? ????? ???? ??? ?????.

???

Kafka? ????? ??? ???? Kafka ????? ????? ???? ?? ??? ???? ?????. ???? ???? ????, ???? ????, Kubernetes? ???? ????, ???? Docker ?? JVM ????? ??? ? ?? ?? ??? ??? ? ????.

???? ??:

• ???? ? ??? ??.
• ????? ??? ?????.
• ??? ? ??? ??.

??? ???? ?? ? ???? ???? ?? ???? ???? ????? ?????.

??

Kafka? ?? ???? ?? ??? ???? ???? ??? ???? ????? ?????. ???? ?? ???? ?? ?????, ?? ???? ??? ??????. ???? ??? ???? ????? ???? ????? ?????.

? ?? ????? ??? ?????.

• ??? ?? ???? ??? ?????? ??? ??
• ?? ??? ??? ??? ?? ?? ???? ?? ??? ???? ?? ?? ??.

Kafka? ??? ???? ????? ???? ??? ?? ??? ?? ?? ??? ??? ??? ?? ? ????.

???

Kafka ????? Kafka ??? ???? ??(??)?? ????? ?????????. Kafka ????? ???(???)? ???? ???? ??? ?????. ???? ??? ??? ?? ?? ??? ?? ???? ??? ?? ? ???? ?????. ???? ??? ?? ??? ??? ?? ????? ????.

• ??? ??:
  • ? ???? ?(?? ??), ?(?? ???) ? ?????? ?????.
  • ?? ???? ???? ???? ??? ?? ?? ???? ??? ?????.
• ??? ??:
  • ?? ???? ???? ?? ????? ???? ???? ?????.
  • ?? ??? ?? ???? ?? ???? ??? ?? ???? ??? ??? ?????.

• ??:

  ???? ???? ???? ???? ???? ?? ?? ???? ?? ? ????.

???

Kafka ???? Kafka ???? ???? ?? Kafka ????? ???? ??? ??? ???? ????? ????????, ???? ??? ?? ?? ??? ?????. . Kafka? ????? ???? ???? ?? ???? ???? Kafka ????? ???? ?????.

???? ??? ??? ???? ?????. ???? ?? ?? ???? ??? ? ???? ???? ?????? ???? ???? ????. ?? ?? ???? ????? ???? ???? ? ??? ??? ?????.

??? ??

??? ??? ?? ??? ???? ???? ?? ???? ??? ????, ?? ???? ?? ??? ?????.

??? ???? ?? ? ??? ?? ?????? ? ???? ?? ? ? ?? ???? ??? ? ????. ?? ??? ??? ?? ?? ??? ??? ????? ??? ? ????.

? ???? ??? ????? ?? ???? ???? Kafka? ?? ???? ????? ?? ? ??? ?? ???? ??????.

??? ? ????

Kafka? ??? ? ????? ?? ? ??? ?? ???? ?? ??? ??? ?? ?? ???? ???? ???? ????? ??? ? ??? ?? ????.

???

??? ??? ??? ????? ??? ? ??? ??? ????? ???? ???????. ???? ???? Kafka ??? ??? ?? ???(?? ?)? ??? ??? ??????.

???? ??? ??:

• JSON: ??? ?? ? ?? ?? ?????.
• Avro: ????? ????? ??? ?????.
• Protobuf: ????? ??? ???? ??? ???? ????.
• ???: ??? ??? ?? ???.
• ??? ?? ???: ??????? ?? ??? ?????.

????

??? ???? ?? ??? ??? ??? ?? ???? ? ???????. ???? Kafka ???? ???? ??? ??? ?? ??? ??? ?? ?? ??? ???? ???????.

??

??? ???? ????? ???? ?? ??? ??? ??? ????. ???, ???, ??? ?? ? ?? ????? ???? ???? ???? ?????, ???? ??? ??? ???, ???? ??? ??????.

???? Kafka ??? ???? ?? ? ?? ?? ???? ??? ? ????. ??? ???? ?? ??? ???? ???? ???? ???? ?? ? ??? ????.

??

• ??? ???? ???: ????? ???? ????? ?? ???? ???? ?? ?????.
• ?? ?? ?? ?? ??: ??? ???? ??? ??? ? ?????.
• ??? ??: ???? ???? ??? ?? ? ?? ??? ?????.

?? ??????

• ?? ??? ??? ?? ??: ???? ??? ??? ?? ?????.
• ???? ???: ???? ? ???? ???? ?? ??? ????.
• ?? ??: ???? ?? ???? ?? ??? ? ??? ?? ? ?????.

??? ???? ???? ????? CPU ???? ?? ?? ?? ??? ???? ?? ??? ??? ??? ???? ?? ?????.

???? ?? ??

• ??: ???? ????(???).
• Gzip: ???? ??? CPU ???? ????.
• Snappy: ?? ?? ?? ??? CPU ????? ??? ?? ??? ?????.
• LZ4: ?? ? ?? ?? ??? ???? ?? ??? ?? ???? ????????.
• Zstd: Gzip?? ??? ??? ?? ???, ?? Kafka ???? ?????.

??

Apache Kafka? ??? ?????? ???? ?? ??? ??? ????? ??? ?? ??? ?? ??? ?? ???? ???. ? ??? ? ??? ???? ??? ?????. Kafka? ??? ? ???? ? ? ?? ??? ??? ????.

• ??? ??:

  • ??? ?: ???? ???? ????? ??? ?? ????. ??? ?? ????? ????? ??? ???? ?????. ??? ??? ??? ???? ?? ??? ?? ?????.

• ??? ??:

  • ?? ??: ???? ??? ?? ??? ????? ?? ?? ??? ???? ?????? Batch.size ? linger.ms? ?????.
  • ??: ??? ??? ??? ???? ? ??? ???? ???? ??(?: ??.??=snappy)? ?????. ???? ?? ?? CPU ????? ????? ?? ?????.

• ??? ??:

  • ???? ??: fetch.min.bytes ? fetch.max.wait.ms? ???? ???? ???? ???? ??? ???? ?????? ?? ??? ?? ?? ??? ???? ??? ????.

?? ??

?? ??? ???? ? ???? ?? ??????? ???? Kafka? ???? ???? ??????? ??? ???. ???? ?? ???? ???? ?? ??? ?????? ??? ???? Kafka ???? ??? ? ????. ? ?????? ?? ??? ??? ? ??? ???? ?????.

{
  temperature: 42,
  timeStamp: new Date(),
};

?? ??? ? ???? ????.

?? Kafka ???? ????? ??? Kafka? ???? ?? ? Docker Kafka ???? ???????.

?? ???? ???? ?????.

docker pull apache/kafka

?? ?? Kafka? ???? ??? ?? ???? ??? ???? ????? ?????.

docker run -d -p 9092:9092 --name Broker apache/kafka:latest

?????. Kafka ???? ?? ????. ???? ?? ??? ???? ???? ??? ???? ?? ??? ??? ? ????. ?? ????? ?? ??? ???? ??? ???? ?? ???.

KafkaJS? ?? NestJS? ???? ??????? ????? ?? Nest CLI? ?? ?????.

nest new my-nest-project

???? ?? ?? kafkajs? ?????

npm? kafkajs

??? ?? ??? ?????

nest g mo kafka

????? ????

nest g mo ???

??? ?? ??

Kafka ??? ??? ??, ?? ??, ?? ? ??? ?? ??? ? ??? ??? ??? ???? ?? Kafka ?? ??? ?????. ?? kafkajs ???? ?? ?? ???? ??? ??? ? ????.

??? ? ??? ??? pub-sub ???(? ?? Kafka)? ??? ?????? ?? ????? ??? ?? ???? ?? ??? ??????.

?? ??? ???? ?????. ?? pub-sub ???? ???? ??? ? ??? ??? ???? ???? ??? ?????.

??? ???? src/interface ??? ??? ? ?? ?? ?????? ??? ?????.

{
  temperature: 42,
  timeStamp: new Date(),
};

// src/interfaces/producer.interface.ts

export interface IProducer {
  produce: (message: any) => Promise<void>;
  connect: () => Promise<void>;
  disconnect: () => Promise<void>;
  isConnected: () => boolean;
}

src/kafka/ ??? ?? ?????? ???? ??? ? ??? ???? ?????.

// src/interfaces/consumer.interface.ts

export type ConsumerMessage = {
  key?: string;
  value: any;
};

export type OnMessage = (message: ConsumerMessage) => Promise<void>;

export interface IConsumer {
  connect: () => Promise<void>;
  disconnect: () => Promise<void>;
  consume: (onMessage?: OnMessage) => Promise<void>;
  isConnected: () => boolean;
}

// src/kafka/kafka.producer.ts

export class KafkaProducer implements IProducer {
  private readonly logger = new Logger(KafkaProducer.name, { timestamp: true });
  private readonly kafka: Kafka;
  private readonly producer: Producer;
  private connected: boolean = false;

  constructor(
    private readonly broker: string,
    private readonly topic: string,
  ) {
    // The client must be configured with at least one broker
    this.kafka = new Kafka({
      brokers: [this.broker],
    });
    this.producer = this.kafka.producer();
  }

  async produce(
    message: Message,
    compression?: CompressionTypes,
    acks?: number,
    timeout?: number,
  ) {
    // To produce, at least a topic and a list of messages must be provided
    await this.producer.send({
      topic: this.topic,
      messages: [message],
      compression,
      timeout,
      acks,
    });
  }

  // To produce a message, the producer must be connected
  async connect() {
    try {
      // Just hooking up some logs in the producer events
      // And storing the connection status
      this.producer.on('producer.connect', () => {
        this.logger.log(
          `producer connected. broker: ${this.broker} topic: ${this.topic}`,
        );
        this.connected = true;
      });

      this.producer.on('producer.disconnect', () => {
        this.logger.log(
          `producer disconnected. broker: ${this.broker} topic: ${this.topic}`,
        );
        this.connected = false;
      });

      // Connect to Kafka
      await this.producer.connect();
    } catch (err) {
      this.logger.error(
        `failed to connect to kafka. broker: ${this.broker} topic: ${this.topic}`,
        err,
      );
    }
  }

  async disconnect() {
    await this.producer.disconnect();
  }

  isConnected(): boolean {
    return this.connected;
  }
}

kafka.module.ts?? ??? ???? ???? ?? ?? ???

// src/kafka/kafka.cosumer.ts

export class KafkaConsumer implements IConsumer {
  private readonly logger = new Logger(KafkaConsumer.name, { timestamp: true });
  private readonly kafka: Kafka;
  private readonly consumer: Consumer;
  private connected: boolean = false;

  constructor(
    private readonly broker: string,
    private readonly topic: string,
    private readonly groupId: string,
  ) {
    if (this.broker && this.topic && this.groupId) {
      // The client must be configured with at least one broker
      this.kafka = new Kafka({
        brokers: [this.broker],
      });
      this.consumer = this.kafka.consumer({ groupId: this.groupId });
    } else {
      this.logger.warn('Broker, topic and groupId must be provided');
    }
  }

  // The onMessage function will be called when a message is received
  async consume(onMessage: OnMessage) {
    // Here we subscribe to the topic ...
    await this.consumer.subscribe({ topic: this.topic });

    // ... and handle the messages
    await this.consumer.run({
      eachMessage: async (payload) => {
        try {
          this.logger.log(
            `message: ${payload.message.value.toString()} (topic: ${payload.topic}, partition: ${payload.partition})`,
          );

          await onMessage({
            key: payload.message.key?.toString(),
            value: payload.message.value.toString(),
          });
        } catch (err) {
          this.logger.error('error on consuming message', err);
        }
      },
    });
  }

  // To consume, the consumer must be connected
  async connect() {
    try {
      // Just hooking up some logs in the consumer events
      // And storing the connection status
      this.consumer.on('consumer.connect', () => {
        this.logger.log(
          `consumer connected. broker: ${this.broker} topic: ${this.topic}`,
        );
        this.connected = true;
      });

      this.consumer.on('consumer.disconnect', () => {
        this.logger.log(
          `consumer disconnected. broker: ${this.broker} topic: ${this.topic}`,
        );
        this.connected = false;
      });

      await this.consumer.connect();
    } catch (err) {
      this.logger.error(
        `failed to connect to kafka. broker: ${this.broker} topic: ${this.topic}`,
        err,
      );
    }
  }

  async disconnect() {
    await this.consumer.disconnect();
  }

  isConnected(): boolean {
    return this.connected;
  }
}

??? ?? ??? ???? Kafka ???? ??????? ??? ??? ? ????. ??? ?? ??? ?? pub-sub ???? ???? ??? ??? ??? ??? ? ?? ????. ??, ?? ???? ???? ??? ?? ? ???? ??? ?? ??? ??? ?/?? ???? ?? ???? ???. ??? ?? ??? ?? pub-sub ????? ????? ?????? ?? ??? ??? ??? ????.

??? ???? ???? ?? Kafka? ??? ? ??? ??? ?? ??? ???? ??? ? ??? ???? ?? ? ????. ?????? ??? ?????.

// src/kafka/kafka.module.ts

@Module({
  imports: [],
  providers: [KafkaProducer, KafkaConsumer],
  exports: [KafkaProducer, KafkaConsumer],
})
export class KafkaModule {}

// src/producer/producer.service.ts

@Injectable()
export class ProducerService implements OnApplicationShutdown {
  // Expects any producer that implements the IProducer interface
  private readonly producer: IProducer;

  constructor(
    @Inject('broker') broker: string,
    @Inject('topic') topic: string,
  ) {
    this.producer = new KafkaProducer(broker, topic);
  }

  /** The produce() and message can receive more parameters,
   * refer to produce method in src/kafka/kafka.producer.ts
   */
  async produce(message: { key?: string; value: string }) {
    if (!this.producer.isConnected()) {
      await this.producer.connect();
    }
    await this.producer.produce(message);
  }

  async onApplicationShutdown() {
    await this.producer.disconnect();
  }
}

?? ????

// src/producer/producer.module.ts

@Module({
  imports: [KafkaModule],
  providers: [
    ProducerService,
    {
      provide: 'broker',
      useValue: 'default-broker-value',
    },
    {
      provide: 'topic',
      useValue: 'default-topic-value',
    },
  ],
  exports: [ProducerService],
})
export class ProducerModule {}

// src/consumer/consumer.service.ts

@Injectable()
export class ConsumerService implements OnApplicationShutdown {
  // Expects any consumer that implements the IConsumer interface
  private readonly consumer: IConsumer;

  constructor(
    @Inject('broker') broker: string,
    @Inject('topic') topic: string,
    @Inject('groupId') groupId: string,
  ) {
    this.consumer = new KafkaConsumer(broker, topic, groupId);
  }

  async consume(onMessage: OnMessage) {
    if (!this.consumer.isConnected()) {
      await this.consumer.connect();
    }
    await this.consumer.consume(onMessage);
  }

  async onApplicationShutdown() {
    await this.consumer.disconnect();
  }
}

?? ?? ?? ??? ??? ? ????. ??.service.ts ???? ??? ???? ???? ?????. ? ???? ???? ???? ???? ?? ???? ?????. ?? ?? ???? ???? ???? ???? ??? ???????.

??? ???? ?? ???? ?????? ??? ? ????. ??? ???? ?? ? ???? ??????? ??? ? onModuleInit ???? ???? ?? ???????.

{
  temperature: 42,
  timeStamp: new Date(),
};

// src/interfaces/producer.interface.ts

export interface IProducer {
  produce: (message: any) => Promise<void>;
  connect: () => Promise<void>;
  disconnect: () => Promise<void>;
  isConnected: () => boolean;
}

?????! Docker ?????? ?? ?? ???? ???? ??????? ???? ???? ??? ?? ? ????. ?? ?? ??? ???? ??? ???? ???? ?? ? ? ????.

docker exec --workdir /opt/kafka/bin/ -it Broker sh

???? ???? ?? ?? ???? ??????? ???? ???, ???? ??, ? ??? ??? ?? ??? ??? ? ????.

? ??? ??? ?? ??????.

? ??? ?? ??? ?? Kafka ?? ??? ?? ?????. ??? ??? PHP ??? ????? ?? ?? ??? ?????!