Benthos框架消息与Ack机制探索

Benthos中Input组件或batchInput组件在Read消息时均会要求返回一个ackfunc, 用于在消息通过output组件后执行ack, 在正常使用的场景下存在以下需要注意的点

1. 无论中间发生了什么, 消息一定会抵达output组件, 在其处理完毕后执行input, 及时中间处理流程返回了非nil的error
2. 对于input组件, 每个消息都会在处理完毕后调用一次跟随消息一起返回的Ack
3. 对于batchInput组件, 将会等待到所返回的那一批消息都处理完毕后, 调用跟随那批消息返回的ack
4. 处理过程中返回的错误或output组件返回的错误都会传递到ackFunc中, 作为参数
5. 消息处理的中间过程不能生成新的消息, 即不能使用service.NewMessage来创建新消息,使用消息自带的Copy来创建新的消息
6. 无论input组件生产了一个消息还是一批消息, 无论中间发生了聚合还是拆分, 只有那次生产的消息都被output处理了, ack才会被执行

针对处理过程中消息的拆分和聚合, 可以做如下验证, 单Input做拆分, Input组件每秒产生一条消息为数字字符串1-100

Processor组件将每条消息拆分成一个或多个, 同时在ackfunc中输出确认的消息, 同时为了保证整个处理串行, 将会限制input组件的max_in_flight为1, 这将会导致系统只允许存在一个未确认消息, 上一个消息没有被确认前, 不会进行Read操作

Benthos 框架下的限速器降级方案探索

rate_limit限速器

用于限制Benthos中并行组件(或跨实例)之间的共享资源使用, 一般使用resources配置, 如下

1
2
3
4
5

rate_limit_resources:  # 固有字段, 表示限速器的资源
  - label: my_limite # 限速器资源标签
    local: # 本地限速器, 只能在单个实例内部生效
      count: 500  # 表示每秒500次处理
      interval: 1s

一些内部组件支持直接在配置中带上rate_limit配置, 例如http_client, 其原理是在组件内部通过*service.Resources的AccessRateLimit方法直接获取对应资源

1
2
3
4
5

input:
  http_client:
    url: TODO
    verb: GET
    rate_limit: my_limite

通过这种方式使用速率限制，可以保证输入仅以每秒 500 个请求的速率轮询 HTTP 源, 其内部实现大致如下

benthos中SetStructured和SetBytes差异

在 Benthos 中，message.SetStructured() 和 message.SetBytes() 是两种不同的消息内容设置方式，它们会影响消息的内部表示形式以及后续处理器的行为。以下是它们的区别和 Benthos 的处理方式：

在 Benthos 的 message 实现中，不能同时有效保存 SetBytes 和 SetStructured 的内容。两种设置方式会互相覆盖，最终生效的是最后一次调用的方法。

• 最后调用的方法决定当前有效形式
  无论先调用哪个方法，​​只有最后一次设置的内容会生效​​，另一种形式会被隐式转换或丢弃。
• 自动转换逻辑
  当访问另一种形式时，Benthos 会按需触发转换

Go语言数据转发处理框架Benthos基础结构及配置

benthos基本配置

benthos的配置文件格式为yaml, 按照逻辑基本可以划分为三个段落, input, processor, output, 数据自input接收, 经过 processor处理, 通过output产出, 另外还有一些logger等其余配置, 各个常用内容大致如下:

• input: 数据的输入源
• processors: 一系列处理器
• caches: 一种键/值存储，可供某些组件用于诸如重复数据删除或数据连接等应用, 一般使用 cache_resources 配置
• rate limits: 限速器, 一般使用 rate_limit_resources 配置
• buffers: 缓冲区, 一般紧接着 input, 用于缓冲输入到下游之间的数据
• metrics: 审计内容
• tracers: 追踪器, 用于消息的跟踪

一个基础的benthos配置文件可以参考如下

Pulsar消费者提交与消费机制小结

消费偏移量与提交偏移量

Pulsar中消费者的消费偏移量（consumption offset）和提交偏移量（committed offset）并不是同一个概念。

1. 消费偏移量（Consumption Offset）
   • 消费偏移量指的是消费者当前正在消费的消息的偏移量。它表示消费者已经从消息流中读取到的位置。
   • 消费偏移量是消费者在消费消息时记录的当前位置，但它并不一定意味着这些消息已经被处理完毕或确认。
2. 提交偏移量（Committed Offset）
   • 提交偏移量是消费者已经成功处理并确认的消息的偏移量。它表示消费者已经处理完这些消息，并且可以安全地认为这些消息不会被再次消费。
   • 提交偏移量通常是通过消费者显式地调用提交偏移量的API来完成的，例如acknowledge或commit方法。

在MIPS及LoongArch的架构下编译go语言二进制

首先需要找到一个已经安装go的机器, 可以是X86或ARM等任何可以执行go的环境

其次准备好go源码

接着进行如下操作

1. 解压go源码

2. 设定环境变量 GOROOT_BOOTSTRAP/GOOS/GOARCH
   其中 GOROOT_BOOTSTRAP 为当前可以执行 go 文件的安装目录, 例如当前环境的 go 可执行文件为/usr/local/go/bin/go, 则 GOROOT_BOOTSTRAP 应设置为 /usr/local/go
GOOS 为目标环境系统
   GOARCH 为目标架构
   关于 GOOS 以及 GOARCH, 可以使用命令, go tool dist list 查看, 版本越高的go支持的越多

信创系统-凝思606替换GLIBC库以运行QT的可行性调查

1. 安装CentOS7编译的带UI的QT程序，其中systemctl相关服务已被替换为service

   由于606系统libc.so.6版本过低，无法正常运行

2. 尝试替换其他高版本的libc.so至本地的lib目录，同时替换其依赖

   

   

枚举

结构体将字段与数据聚合，而枚举可以将同一类型的东西（同时你可以将它可能的情况列举出来）作为一个集合，使用IP地址举例如下

1
2
3
4

enum IpAddrKind {
	v4,
	v6,
}

枚举值

可使用::访问枚举值

1
2

let ipv4 = IpAddrKind::v4;
let ipv6 = IpAddrKind::v6;

也可将其传入函数，如func(IpAddrKind::v4)

结构体的定义与实例化

举例如下

1
2
3
4
5
6

struct User {
	active: bool,
	username: String,
	email: String,
	sign_in_count: u64,
}

实例化如下

1
2
3
4
5
6
7
8
9
10
11
12

let user1 = User {
	active: true,
	username: String::from("name"),
	email: String::from("email@email.com"),
	sign_in_count: 1,
};
//若要使用值，则同C/C++
//若要修改值，则示例需是可变的，如下
let mut user2 = User {
    //....
}
user2.email = String::from("xxx");

Rust 通过所有权系统管理内存，编译器在编译时会根据一系列规则进行检查，如果违反了这些规则，程序不能通过编译。在运行时，所有权系统的任何功能都不会减缓程序的运行

所有权规则

1. Rust 中的每一个值都有一个所有者
2. 值在任意时刻有且仅有一个所有者
3. 当所有者（变量）离开作用域，这个值将被丢弃