首页 / 消息队列 / KOA+egg.js集成kafka消息队列的示例

KOA+egg.js集成kafka消息队列的示例

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了KOA+egg.js集成kafka消息队列的示例，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含7309字，纯文字阅读大概需要11分钟。

内容图文

Egg.js ：基于KOA2的企业级框架

Kafka：高吞吐量的分布式发布订阅消息系统

本文章将集成egg + kafka + mysql 的日志系统例子

系统要求：日志记录，通过kafka进行消息队列控制

思路图：

KOA+egg.js集成kafka消息队列的示例 - 文章图片

这里消费者和生产者都由日志系统提供

λ.1 环境准备

①Kafka

官网下载kafka后，解压

启动zookeeper：

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

启动Kafka server

这里config/server.properties中将num.partitions=5，我们设置5个partitions

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

② egg + mysql

根据脚手架搭建好egg，再多安装kafka-node，egg-mysql

mysql 用户名root 密码123456

λ.2 集成

1、根目录新建app.js，这个文件在每次项目加载时候都会运作

'use strict';
 
const kafka = require('kafka-node');
 
module.exports = app => {
 app.beforeStart(async () => {
 const ctx = app.createAnonymousContext();
 
 const Producer = kafka.Producer;
 const client = new kafka.KafkaClient({ kafkaHost: app.config.kafkaHost });
 const producer = new Producer(client, app.config.producerConfig);
 
 producer.on('error', function(err) {
  console.error('ERROR: [Producer] ' + err);
 });
 
 app.producer = producer;
 
 const consumer = new kafka.Consumer(client, app.config.consumerTopics, {
  autoCommit: false,
 });
 
 consumer.on('message', async function(message) {
  try {
  await ctx.service.log.insert(JSON.parse(message.value));
  consumer.commit(true, (err, data) => {
   console.error('commit:', err, data);
  });
  } catch (error) {
  console.error('ERROR: [GetMessage] ', message, error);
  }
 });
 
 consumer.on('error', function(err) {
  console.error('ERROR: [Consumer] ' + err);
 });
 });
};

上述代码新建了生产者、消费者。

生产者新建后加载进app全局对象。我们将在请求时候生产消息。这里只是先新建实例

消费者获取消息将访问service层的insert方法（数据库插入数据）。

具体参数可以参考kafka-node官方API，往下看会有生产者和消费者的配置参数。

2、controller · log.js

这里获取到了producer，并传往service层

'use strict';
 
const Controller = require('egg').Controller;
 
class LogController extends Controller {
 /**
 * @description Kafka控制日志信息流
 * @host /log/notice
 * @method POST
 * @param {Log} log 日志信息
 */
 async notice() {
 const producer = this.ctx.app.producer;
 const Response = new this.ctx.app.Response();
 
 const requestBody = this.ctx.request.body;
 const backInfo = await this.ctx.service.log.send(producer, requestBody);
 this.ctx.body = Response.success(backInfo);
 }
}
 
module.exports = LogController;

3、service · log.js

这里有一个send方法，这里调用了producer.send ，进行生产者生产

insert方法则是数据库插入数据

'use strict';
 
const Service = require('egg').Service;
const uuidv1 = require('uuid/v1');
 
class LogService extends Service {
 async send(producer, params) {
 const payloads = [
  {
  topic: this.ctx.app.config.topic,
  messages: JSON.stringify(params),
  },
 ];
 
 producer.send(payloads, function(err, data) {
  console.log('send : ', data);
 });
 
 return 'success';
 }
 async insert(message) {
 try {
  const logDB = this.ctx.app.mysql.get('log');
  const ip = this.ctx.ip;
 
  const Logs = this.ctx.model.Log.build({
  id: uuidv1(),
  type: message.type || '',
  level: message.level || 0,
  operator: message.operator || '',
  content: message.content || '',
  ip,
  user_agent: message.user_agent || '',
  error_stack: message.error_stack || '',
  url: message.url || '',
  request: message.request || '',
  response: message.response || '',
  created_at: new Date(),
  updated_at: new Date(),
  });
 
  const result = await logDB.insert('logs', Logs.dataValues);
 
  if (result.affectedRows === 1) {
  console.log(`SUCEESS: [Insert ${message.type}]`);
  } else console.error('ERROR: [Insert DB] ', result);
 } catch (error) {
  console.error('ERROR: [Insert] ', message, error);
 }
 }
}
 
module.exports = LogService;

4、config · config.default.js

一些上述代码用到的配置参数具体在这里，注这里开了5个partition。

'use strict';
 
module.exports = appInfo => {
 const config = (exports = {});
 
 const topic = 'logAction_p5';
 
 // add your config here
 config.middleware = [];
 
 config.security = {
 csrf: {
  enable: false,
 },
 };
 
 // mysql database configuration
 config.mysql = {
 clients: {
  basic: {
  host: 'localhost',
  port: '3306',
  user: 'root',
  password: '123456',
  database: 'merchants_basic',
  },
  log: {
  host: 'localhost',
  port: '3306',
  user: 'root',
  password: '123456',
  database: 'merchants_log',
  },
 },
 default: {},
 app: true,
 agent: false,
 };
 
 // sequelize config
 config.sequelize = {
 dialect: 'mysql',
 database: 'merchants_log',
 host: 'localhost',
 port: '3306',
 username: 'root',
 password: '123456',
 dialectOptions: {
  requestTimeout: 999999,
 },
 pool: {
  acquire: 999999,
 },
 };
 
 // kafka config
 config.kafkaHost = 'localhost:9092';
 
 config.topic = topic;
 
 config.producerConfig = {
 // Partitioner type (default = 0, random = 1, cyclic = 2, keyed = 3, custom = 4), default 0
 partitionerType: 1,
 };
 
 config.consumerTopics = [
 { topic, partition: 0 },
 { topic, partition: 1 },
 { topic, partition: 2 },
 { topic, partition: 3 },
 { topic, partition: 4 },
 ];
 
 return config;
};

5、实体类：

mode · log.js

这里使用了 Sequelize

'use strict';
 
module.exports = app => {
 const { STRING, INTEGER, DATE, TEXT } = app.Sequelize;
 
 const Log = app.model.define('log', {
 /**
  * UUID
  */
 id: { type: STRING(36), primaryKey: true },
 /**
  * 日志类型
  */
 type: STRING(100),
 /**
  * 优先等级(数字越高，优先级越高）
  */
 level: INTEGER,
 /**
  * 操作者
  */
 operator: STRING(50),
 /**
  * 日志内容
  */
 content: TEXT,
 /**
  * IP
  */
 ip: STRING(36),
 /**
  * 当前用户代理信息
  */
 user_agent: STRING(150),
 /**
  * 错误堆栈
  */
 error_stack: TEXT,
 /**
  * URL
  */
 url: STRING(255),
 /**
  * 请求对象
  */
 request: TEXT,
 /**
  * 响应对象
  */
 response: TEXT,
 /**
  * 创建时间
  */
 created_at: DATE,
 /**
  * 更新时间
  */
 updated_at: DATE,
 });
 
 return Log;
};

6、测试Python脚本：

import requests
 
from multiprocessing import Pool
from threading import Thread
 
from multiprocessing import Process
 
 
def loop():
 t = 1000
 while t:
  url = "http://localhost:7001/log/notice"
 
  payload = "{\n\t\"type\": \"ERROR\",\n\t\"level\": 1,\n\t\"content\": \"URL send ERROR\",\n\t\"operator\": \"Knove\"\n}"
  headers = {
  'Content-Type': "application/json",
  'Cache-Control': "no-cache"
  }
 
  response = requests.request("POST", url, data=payload, headers=headers)
 
  print(response.text)
 
if __name__ == '__main__':
 for i in range(10):
  t = Thread(target=loop)
  t.start()

7、建表语句：

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
 
-- ----------------------------
-- Table structure for logs
-- ----------------------------
DROP TABLE IF EXISTS `logs`;
CREATE TABLE `logs` (
 `id` varchar(36) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
 `type` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '日志类型',
 `level` int(11) NULL DEFAULT NULL COMMENT '优先等级(数字越高，优先级越高）',
 `operator` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL DEFAULT NULL COMMENT '操作人',
 `content` text CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '日志信息',
 `ip` varchar(36) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL DEFAULT NULL COMMENT 'IP\r\nIP',
 `user_agent` varchar(150) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL DEFAULT NULL COMMENT '当前用户代理信息',
 `error_stack` text CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '错误堆栈',
 `url` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL DEFAULT NULL COMMENT '当前URL',
 `request` text CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '请求对象',
 `response` text CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '响应对象',
 `created_at` datetime(0) NULL DEFAULT NULL COMMENT '创建时间',
 `updated_at` datetime(0) NULL DEFAULT NULL COMMENT '更新时间',
 PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_bin ROW_FORMAT = Dynamic;
 
SET FOREIGN_KEY_CHECKS = 1;

λ.3 后话

网上类似资料甚少，啃各种文档，探寻技术实现方式

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

内容总结

以上是互联网集市为您收集整理的KOA+egg.js集成kafka消息队列的示例全部内容，希望文章能够帮你解决KOA+egg.js集成kafka消息队列的示例所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/334198.html

来源：【匿名】

【上一篇】.NetCore利用BlockingCollection实现简易消息队列【下一篇】PHP 消息队列 Kafka 使用

更多 ►

【KOA+egg.js集成kafka消息队列的示例】教程文章相关的互联网学习教程文章

大数据安装之Kafka（用于实时处理的消息队列）【图】

一、安装部署kafka1、集群规划hadoop102 hadoop103 hadoop104zk zk zkkafka kafka kafka2、jar包下载http://kafka.apache.org/downloads.html 3、集群部署1）解压安装包[test@hadoop102 software...

kafka——分布式的消息队列系统

总听公司人说kafka kafka...所以这玩意到底是个啥？好像是一个高级版的消息队列，什么高吞吐量，数据持久，消息均衡，emmmhttps://blog.csdn.net/nawenqiang/article/details/81705276原文：https://www.cnblogs.com/gaoquanquan/p/10809977.html

一、kafka一个最基本的架构认识　　由多个broker组成，每个broker就是一个节点；创建一个topic，这个topic可以划分为多个partition，每个partition可以存在于不同的broker上，每个partition存放放一部分数据。　　kafka就是一个分布式消息队列，就是说一个topic的数据，是分散放在多个机器上的，每个机器就放一部分数据。　　实际上像rabbitmq之类的，并不是分布式消息队列，它只是传统的消息队列，只不过提供了一些集群、HA的机制...

面试知识点总结之RabbitMQ/Kafka使用场景

RabbitMQ消息丢失的情况 provider->Rabbitmq->consumer 1. 消息在传送过程中丢失方案①开启RabbitMQ事务（同步，不推荐） ②开启Confim模式（异步，推荐） 2. Rabbitmq收到消息，暂存内存中，还没消费，自己挂掉了，内存中的数据丢失方案：开启RabbitMQ持久化 3. 消费者消费到了这个消息，但还没来得及处理就挂了，Rabbitmq以为消息已经被处理了方案：关闭RabbitMQ自动ACK kafka消息丢失和重复消费 2两面分析一、发送方式：...

RabbitMQ、Kafka、RocketMQ 是如何实现高可用的？【图】

本文从高可用的角度观察一下 RabbitMQ、Kafka、RocketMQ，看看它们各自的实现思路。 1. RabbitMQ RabbitMQ 有 3 种部署模式：单机模式普通集群模式镜像集群模式单机模式与高可用完全没关系，咱就不说了，直接看看这2种集群模式。 1.1 普通集群模式某一个 Queue 是在集群中的某一个 Broker 上，各个 Broker 会同步元数据，但不会同步 Queue 的消息数据。如果某一个 Broker 故障了，其中的 Queue 便无法使用。如果消息没有配置消息...

从17 个方面对比 Kafka、RabbitMQ、RocketMQ、ActiveMQ 等分布式消息队列

本文转载于： https://www.zhihu.com/question/43557507 知乎专题： Kafka，Mq，Redis作为消息队列使用时的差异？（里面还有很多回答值得去看和思考） 1 资料文档 Kafka：中。有 kafka 作者自己写的书，网上资料也有一些。 rabbitmq：多。有一些不错的书，网上资料多。 zeromq：少。没有专门写 zeromq 的书，网上的资料多是一些代码的实现和简单介绍。 rocketmq：少。没有专门写 rocketmq 的书，网上的资料良莠不齐，官...

RabbitMQ和Kafka的比较 - 再也不要把它们混为一谈【图】

经常有人会问: “应该选择RabbitMQ还是Kafka？”。基于某些原因，许多开发者会把这两种技术当做等价的来看待。的确，在一些案例场景下选择RabbitMQ还是Kafka没什么差别，但是这两种技术在底层实现方面是有许多差异的。不同的场景需要不同的解决方案，选错一个方案能够严重的影响你对软件的设计，开发和维护的能力。异步消息模式异步消息可以作为解耦消息的生产和处理的一种解决方案。提到消息系统，我们通常会想到两种主要...

关于rabbitmq与kafka的异同

场景：对于同一份数据，多个业务消费。 rabbitmq：交换机根据订阅者以及订阅者设置的队列，分发数据到对应业务的queue里面，各自消费队列不影响。 kafka：公用同一份数据，不同的订阅者根据位点，请求数据。相同的groupid的消费者，公用同一个位点。

windows下启动和运行分布式消息中间件消息队列 kafka【代码】

本文转载至:https://www.cnblogs.com/flower1990/p/7466882.html 一、安装JAVA JDK 1、下载安装包 http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html 注意：根据32/64位操作系统下载对应的安装包 2、添加系统变量：JAVA_HOME=C:\Program Files (x86)\Java\jdk1.8.0_144 二、安装ZooKeeper 1、下载安装包 http://zookeeper.apache.org/releases.html#download 2、解压并进入ZooKeeper目录，笔...

PHP 消息队列 Kafka 使用【代码】【图】

安装 Kafka 服务直接到 kafka 官网 , 下载最新的wget https://mirror.bit.edu.cn/apache/kafka/2.5.0/kafka_2.13-2.5.0.tgz解压，进入目录tar -zxvf kafka_2.13-2.5.0.tgz cd kafka_2.13-2.5.0启动 Kafka 服务使用安装包中的脚本启动单节点 Zookeeper 实例bin/zookeeper-server-start.sh -daemon config/zookeeper.properties使用 kafka-server-start.sh 启动 kafka 服务bin/kafka-server-start.sh config/server.properties创建 t...