go-zero(十二)消息队列

news2025/2/27 20:43:52

go zero 消息队列

在微服务架构中，消息队列主要通过异步通信实现服务间的解耦，使得各个服务可以独立发展和扩展。

go-zero中使用的队列组件go-queue，是gozero官方实现的基于Kafka和Beanstalkd 的消息队列框架,我们使用kafka作为演示。

一、kafka简单介绍

Kafka 是一个开源的分布式流处理平台，主要用于构建实时数据管道和流应用。

1.Kafka 的架构

Kafka 的架构通常由以下几个部分构成：

  • Broker(节点)：Kafka 集群由多个 broker 实例组成，负责管理消息的存储和处理。
  • Topic(主题)：消息以主题的形式组织，每个主题可以有多个分区（partition），以支持高并发和扩展性。
  • Produce(生产者)：消息生产者将数据发送到特定的topic中。
  • Consumer(消费者)：消费者从topic中读取数据，可以将多个消费者分组以进行负载均衡。

2.Kafka 的关键特性

  1. 高吞吐量：
    Kafka 设计上能够处理大量的实时数据流，具备非常高的吞吐量。这使得它能够轻松应对大规模数据流量，适合做日志聚合、监控数据处理等。

  2. 持久性：
    Kafka 将消息持久化到磁盘，并提供复制功能，以确保数据的安全性和可靠性。即使在节点出现故障的情况下，也能保证数据不会丢失。

  3. 可扩展性：
    Kafka 能够水平扩展，通过增加更多的节点来处理更多的消费者和生产者，这使得它能够应对越来越多的业务需求。

  4. 实时处理：
    Kafka 提供低延迟的数据传输，这使得实时处理和分析成为可能。您可以瞬时处理到来的数据流。

  5. 支持多种消息传递模式：
    Kafka 支持发布-订阅和点对点的消息传递模式，能够灵活适应不同场景下的需求。

  6. 强大的生态系统：
    Kafka 拥有丰富的生态系统，包括 Kafka Streams 和 Kafka Connect，这些工具可以帮助开发者更方便地进行流处理和数据集成。

3.常见应用场景

  1. 日志聚合：
    Kafka 可以作为一个集中式的日志聚合器，将分布在不同服务的日志集中到一个地方，方便后续分析和监控。

  2. 实时数据流处理：
    使用 Kafka，用户可以实时处理和分析流数据，例如检测异常、生成实时报告等。

  3. 系统监控和事件追踪：
    Kafka 经常用于收集和跟踪系统事件（如用户行为、系统状态等），并通过流式处理进行实时监控。

  4. 数据集成：
    Kafka 可以作为数据的桥梁，连接不同的数据源和目标系统，方便实现数据的流转和转换。

  5. 消息队列：
    Kafka 可用作高效的消息队列，实现服务间的异步通信。例如，在微服务架构中，服务 A 可以将消息发送到 Kafka，而服务 B 可以异步地从 Kafka 中读取处理这些消息。

二、环境部署

1.Docker安装Kafka

配置文件：

version: '3'

######## 项目依赖的环境，启动项目之前要先启动此环境 #######

services:

  #zookeeper是kafka的依赖 - Zookeeper is the dependencies of Kafka
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    environment:
      # 时区上海 - Time zone Shanghai (Change if needed)
      TZ: Asia/Shanghai
    restart: always
    ports:
      - 2181:2181
    networks:
      - gozero_net

  #消息队列 - Message queue
  kafka:
    image: 'bitnami/kafka:3.6.2'
    container_name: kafka
    restart: always
    ulimits:
      nofile:
        soft: 65536
        hard: 65536
    environment:
      - TZ=Asia/Shanghai
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
    ports:
      - '9092:9092'
      - '9094:9094'
    volumes:
      - ./volumes/kafka:/bitnami/kafka


networks:
  gozero_net:
    driver: bridge
    ipam:
      config:
        - subnet: 172.16.0.0/16

这里的kafka 对外暴露的端口为9094

使用命令执行文件

docker compose up

2.创建topic

进入kafka容器

docker exec -it {{容器ID}} /bin/bash
#或者直接使用容器名
docker exec -it kafka /bin/bash

进入kafka执行命令目录
进入kafka执行命令目录

cd /opt/bitnami/kafka/bin

创建topic

创建名为topic-test的topic

./kafka-topics.sh --create --topic topic-test --bootstrap-server localhost:9092

查看topic信息

./kafka-topics.sh --describe --topic topic-test --bootstrap-server localhost:9092

3.测试topic

使用两个终端，进入kafka执行命令目录，执行下面两个命令：

生产消息

./kafka-console-producer.sh --topic topic-test --bootstrap-server localhost:9092

消费消息

./kafka-console-consumer.sh --topic topic-test --bootstrap-server localhost:9092

测试

在生产者输入消息，会自动同步到消费者
在这里插入图片描述

4. 拉取依赖

项目中首先要拉取 go-queue 的依赖

go get github.com/zeromicro/go-queue@latest

三、项目演示

1.配置说明

type KqConf struct {
   service.ServiceConf

   // Brokers: Kafka 的多个 Broker 节点
   Brokers []string

   // Group: 消费者组
   Group string

   // Topic: 订阅的 Topic 主题
   Topic string

   // Offset: 如果新的 topic Kafka 没有对应的 offset 信息，或者当前的 offset 无效（历史数据被删除），
   // 需要指定从头（first）消费还是从尾（last）消费
   Offset string `json:",options=first|last,default=last"`

   // Conns: 一个 Kafka queue 对应可对应多个 consumer，Conns 对应 Kafka queue 数量，
   // 可以同时初始化多个 Kafka queue，默认只启动一个
   Conns int `json:",default=1"`

   // Consumers: go-queue 内部起多个 goroutine 从 Kafka 中获取信息写入进程内的 channel，
   // 此参数控制 goroutine 数量（⚠️ 并不是真正消费时的并发 goroutine 数量）
   Consumers int `json:",default=8"`

   // Processors: 当 Consumers 中的多个 goroutine 拉取到 Kafka 消息后，
   // 通过此参数控制当前消费逻辑的并发 goroutine 数量
   Processors int `json:",default=8"`

   // MinBytes: fetch 一次返回的最小字节数，若不够该字节数则会等待
   MinBytes int `json:",default=10240"`    // 10K

   // MaxBytes: fetch 一次返回的最大字节数，若第一条消息大小超过该限制仍会继续拉取，
   // 以确保 consumer 的正常运行。并非绝对配置，消息大小也受 broker 的 message.max.bytes 限制，
   // 以及 topic 的 max.message.bytes 限制
   MaxBytes int `json:",default=10485760"` // 10M

   // Username: Kafka 的账号（可选）
   Username string `json:",optional"`

   // Password: Kafka 的密码（可选）
   Password string `json:",optional"`
}

2.配置

配置文件

在yaml 配置文件中添加当前的 kafka 配置信息，我这里为了省事就都放在一个配置文件下了：

#....

#生产者
KqPusherConf:
  Name: log-producer
  Brokers:
    - 127.0.0.1:9094
  Group: logs-group
  Topic: topic-test
#消费者
KqConsumerConf:
    Name: log-consumer
    Brokers:
      - 127.0.0.1:9094
    Group: logs-group
    Topic: topic-test
    Offset: last
    Consumers: 8
    Processors: 8

config.go

在 internal/config 下的 config.go 中定义 go 映射的配置


type Config struct {

	/*
	.....
	*/
	KqPusherConf   kq.KqConf
	KqConsumerConf kq.KqConf
}

svc注入
在 svc/serviceContext.go 中初始化 pusher 的 kq client

type ServiceContext struct {
	Config  config.Config
	KqPusherClient *kq.Pusher
	
}

func NewServiceContext(c config.Config) *ServiceContext {
	return &ServiceContext{
		Config:  c,
		KqPusherClient: kq.NewPusher(c.KqPusherConf.Brokers, c.KqPusherConf.Topic),
	}
}

3. 生产者

在 logic 中写业务逻辑使用 go-queue 的 kq client 发送消息到 kafka，这里我们用登录作为演示，当登录成功后，发送用户信息：


func (l *LoginLogic) Login(req *types.LoginRequest) (resp *types.LoginResponse, err error) {
	// todo: add your logic here and delete this line
	
	/*
	....省略其他代码
	*/
	
	//生产者需要异步执行，threading.GoSafe() 是go zero官方对 go func() 的安全封装
	threading.GoSafe(func() {
		
		logData := map[string]any{
			"user":   user.Username,
			"mobile": user.Mobile,
		}
		logs, _ := json.Marshal(logData)
		// 使用Push推送消息，消息为json
		err := l.svcCtx.KqPusherClient.Push(l.ctx, string(logs))
		if err != nil {
			logx.Errorf("KqPusherClient Push Error , err :%v", err)
		}
	})

	// 如果既没有验证码也没有密码
	return nil, errors.New(10010, "未提供有效的登录凭证")	
}

生产者需要异步执行，threading.GoSafe() 是go zero官方对 go func() 的安全封装，如果出现panics 会自动恢复。

4. 消费者

在 internal 下新建一个 mq 文件夹，在 mq 文件夹下新建一个消费者 consumer.go:

package mqs

import (
	"beyond/user/api/internal/svc"
	"context"
	"fmt"
	"github.com/zeromicro/go-zero/core/logc"
	"github.com/zeromicro/go-zero/core/logx"
)

//定义日志消费者
type LogsConsumer struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

// 定义构造方法
func NewLogsConsumer(ctx context.Context, svcCtx *svc.ServiceContext) *LogsConsumer {
	return &LogsConsumer{
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

// Consume为go zero内置接口， 实现Consume接口方法
func (l *LogsConsumer) Consume(ctx context.Context, key, val string) error {
	
	//logx.Infof("Consumer key :%s , val :%s", key, val)
	logc.Infof(ctx, "Consumer key :%s, val :%s", key, val)
	return nil
}

Consume 为go queue内置接口
在这里插入图片描述

因为消费者可能有多个，在 mq 文件夹下新建一个文件mqs.go用来监听多个消费者,mqs.go 代码如下:

package mqs

import (
	"beyond/user/api/internal/config"
	"beyond/user/api/internal/svc"
	"context"

	"github.com/zeromicro/go-queue/kq"
	"github.com/zeromicro/go-zero/core/service"
)

func Consumers(c config.Config, ctx context.Context, svcCtx *svc.ServiceContext) []service.Service {
	// 监听消费者状态变化
	return []service.Service{
		//创建消息队列
		kq.MustNewQueue(c.KqConsumerConf, NewLogsConsumer(ctx, svcCtx)),
	}

}

在 main.go 中启动 consumers 等待消费


func main() {
	flag.Parse()

	var c config.Config
	conf.MustLoad(*configFile, &c)

	server := rest.MustNewServer(c.RestConf)
	defer server.Stop()

	svcCtx := svc.NewServiceContext(c)
	handler.RegisterHandlers(server, svcCtx)

	fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
	// 因为现在添加了mq,属于多服务状态,所以需要启动mq服务
	//server.Start()  
	//创建新的服务组
	serviceGroup := service.NewServiceGroup()
	defer serviceGroup.Stop()
	// 从mq中获取消费者服务，并添加到服务组中
	for _, mq := range mqs.Consumers(c, context.Background(), svcCtx) {
		serviceGroup.Add(mq)
	}
	//添加原来的server服务
	serviceGroup.Add(server)
	// 启动服务组
	serviceGroup.Start()

}

5.启动项目

我们这里就是简单的输出日志，你也可以拓展成发送邮件或者短信给用户，提示用户注册成功。
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2257969.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

day30-IO-阶段综合案例（带权重的随机每日一记）（笔记完全来源于黑马程序员）

day30-IO-阶段综合案例（带权重的随机每日一记）（笔记完全来源于黑马程序员）

目录 0 目录一、听黑马阿玮的视频记录的笔记1. 制造假数据1.1 如何制造假数据1.2 练习1-生成方式1：爬取姓氏、男生名字、女生名字1.3 练习2-生成方式1：在练习1的基础上，将数据写入本地文件1.4 练习3-生成方式2：利用糊涂包生成假数…
阅读更多...
Excel的文件导入遇到大文件时

Excel的文件导入遇到大文件时

Excel的文件导入向导如何把已导入数据排除 入起始行，选择从哪一行开始导入。 比如，前两行已经导入了，第二次导入的时候排除前两行，从第三行开始，就将导入起始行设置为3即可，且不勾选含标题行。 但遇到大文…
阅读更多...
qemu安装arm64架构银河麒麟

qemu安装arm64架构银河麒麟

qemu虚拟化软件，可以在一个平台上模拟另一个硬件平台，可以支持多种处理器架构。 一、安装 安装教程：https://blog.csdn.net/qq_36035382/article/details/125308044 下载链接：https://qemu.weilnetz.de/w64/2024/ 我下载的是 …
阅读更多...
Linux-ubuntu环境配置

Linux-ubuntu环境配置

一，安装VWware，里面导入镜像文件 这些都是文件夹里面有的，然后对着正点原子视频安装就行，虚拟机的破解码，去百度搜一个能用就行，中间遇见俩问题。①乌班图里面不能上网，②插入U盘后，…
阅读更多...
如何在 Ubuntu 22.04 上安装和使用 Apache Kafka

如何在 Ubuntu 22.04 上安装和使用 Apache Kafka

简介 Apache Kafka是一个高性能、低延迟的分布式流处理平台，广泛用于构建实时数据管道和流式应用。本文将指导你如何在Ubuntu 22.04系统上快速部署Apache Kafka，让你体验到Kafka在处理大规模实时数据流方面的强大能力。通过本教程，你将学会如…
阅读更多...
TCP/IP杂记

TCP/IP杂记

TCP三次握手、四次挥手 从应用角度，不用多考虑为什么有三次，遵循标准即可。 ubuntu 下 wireshark安装： sudo add-apt-repository universe sudo apt install wireshark 三次握手实证： 第一次握手的情况如下：&#…
阅读更多...
【2024最新Java面试宝典】—— SpringBoot面试题（44道含答案）

【2024最新Java面试宝典】—— SpringBoot面试题（44道含答案）

1. 什么是 Spring Boot？ Spring Boot 是 Spring 开源组织下的子项目，是 Spring 组件一站式解决方案，主要是简化了使用 Spring 的难度，简省了繁重的配置，提供了各种启动器，使开发者能快速上手。 2. 为什么…
阅读更多...
MTK Android12 开机向导

MTK Android12 开机向导

文章目录 需求-场景参考资料：博客资料官网参考资料：参考资料注意点 附件资料文件说明：推荐工具：配置定制的 声明叠加层 APK需求实现替换字符、删减开机向导界面、添加开机向导界面删除部分界面需求，官网说明如下更新部…
阅读更多...
思科模拟器路由器的基本配置

思科模拟器路由器的基本配置

一、实验目的 了解路由器的作用掌握路由器的基本配置方法 3、掌握路由器模块的使用和互连方式 二、实验环境 2811路由器一台，计算机两台，Console配置线一根，网线若干；本实验拓扑图如图8-1所示；计算机IP地址规划如表8-…
阅读更多...
HTTP multipart/form-data 请求

HTTP multipart/form-data 请求

序言 最近在写项目的过程中有一个需求是利用 HTTP 协议传输图片和视频，经过查询方法相应的方法发现使用 multipart/form-data 的方式，这是最常见处理二进制文件的表单编码类型。  学习了一下午，现在总结一下使用的方法和相关的知识点&#x…
阅读更多...
一、测试工具LoadRunner Professional脚本编写-录制前设置

一、测试工具LoadRunner Professional脚本编写-录制前设置

设置基于URL的脚本 原因：基于HTML的脚本会导致login接口不能正确录制 设置UTF-8 原因：不勾选此项会导致脚本中文变为乱码
阅读更多...
Hadoop生态圈框架部署（九-2）- Hive HA（高可用）部署

Hadoop生态圈框架部署（九-2）- Hive HA（高可用）部署

文章目录 前言一、Hive部署（手动部署）下载Hive1. 上传安装包2. 解压Hive安装包2.1 解压2.2 重命名2.3 解决冲突2.3.1 解决guava冲突2.3.2 解决SLF4J冲突 3. 配置Hive3.1 配置Hive环境变量3.2 修改 hive-site.xml 配置文件3.3 配置MySQL驱动包3.3.1 下在M…
阅读更多...
JAVA ｜日常开发中Websocket详解

JAVA ｜日常开发中Websocket详解

JAVA ｜日常开发中Websocket详解 前言一、Websocket 概述1.1 定义1.2 优势 二、Websocket 协议基础2.1 握手过程2.2 消息格式2.3 数据传输方式 三、Java 中使用 Websocket3.1 Java WebSocket API（JSR - 356）3.2 第三方库（如 Tyrus&…
阅读更多...
算法基础学习Day6(动态窗口)

算法基础学习Day6(动态窗口)

文章目录 1.题目2.题目解答1.最大连续1的个数题目及题目解析算法学习思路一:暴力解法思路二:滑动窗口 代码提交 2.将x减到0的最小操作数题目及题目解析算法学习滑动窗口解决问题 代码提交 1.题目 1004. 最大连续1的个数 III - 力扣（LeetCode）1658. 将 x…
阅读更多...
开发一套SDK 第一弹

开发一套SDK 第一弹

自动安装依赖包 添加条件使能 #ex: filetypesh bash_ls 识别 达到预期,多个硬件环境 等待文件文件系统挂在完成 或者创建 /sys/class/ 属性文件灌入配置操作 AI 提供的 netlink 调试方法,也是目前主流调用方法,socket yyds #include <linux/module.h> #include <linux…
阅读更多...
【leetcode】替换后的最长重复字符、将字符串翻转到单调递增

【leetcode】替换后的最长重复字符、将字符串翻转到单调递增

1.替换后的最长重复字符 示例如下&#xff1a; 下面我们来分析一下一个例子&#xff0c;其中K 2 暴力枚举 这里的字符串s是仅由大写字母组成&#xff0c;首先我们尝试用暴力解法的思路来想一下这道题&#xff0c;通过从第一个字符开始进行枚举&#xff0c;如果出现了条件判断…
阅读更多...
HarmonyOS 线性容器List 常用的几个方法

HarmonyOS 线性容器List 常用的几个方法

List底层通过单向链表实现&#xff0c;每个节点有一个指向后一个元素的引用。当需要查询元素时&#xff0c;必须从头遍历&#xff0c;插入、删除效率高&#xff0c;查询效率低。List允许元素为null。 List和LinkedList相比&#xff0c;LinkedList是双向链表&#xff0c;可以快速…
阅读更多...
C# 网络编程--基础核心内容

C# 网络编程--基础核心内容

在现今软件开发中&#xff0c;网络编程是非常重要的一部分&#xff0c;本文简要介绍下网络编程的概念和实践。 C#网络编程的主要内容包括以下几个方面‌&#xff1a; : 上图引用大佬的图&#xff0c;大家也关注一下&#xff0c;有技术有品质&#xff0c;有国有家&#xff0c;情…
阅读更多...
基于Python实现web网页内容爬取

基于Python实现web网页内容爬取

文章目录 1. 网页分析2. 获取网页信息2.1 使用默认的urllib.request库2.2 使用requests库1.3 urllib.request 和 requests库区别 2. 更改用户代理3. BeautifulSoup库筛选数据3.1 soup.find()和soup.find_all() 函数 4. 抓取分页链接参考资料 在日常学习和工作中&#xff0c;我们…
阅读更多...
ASP .NET Core 中的环境变量

ASP .NET Core 中的环境变量

在本文中&#xff0c;我们将通过组织一场小型音乐会&#xff08;当然是在代码中&#xff09;来了解 ASP .NET Core 中的环境变量。让我们从创建项目开始&#xff1a; dotnet new web --name Concert 并更新Program.cs&#xff1a; // replace this: app.MapGet("/"…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023