nsqio

nsq

nsq 支持分布式事务(四)

也许我就是个跑偏的人,这个系列越写越偏 基准测试 本来是跟原版里面./bench.sh做一些基准测试,在我这台垃圾电脑上面 写性能 最后发现如果没有SyncTimout(就是超时落盘)这个,dtnsq写竟然40w/s,而nsq才18w/s~20w/s 没办法加上了默认了SyncTimout(2s),但是dtnsq也是30w/s 大家在落盘之前都是内存操作,如果有性能差异,我猜测是 channel导致的 读性能 读性能没法比较,因为当前dtnsq根本没有做push,只是定时拉取 修改 我需要将一些代码细节优化(不是为了写性能),稍微向生产方向靠近再来进一步测试 优化定时器 之前的消费者的pull定时器是从nsqd里面触发的,很容易将整个nsqd``lock住 这里面改成每一个读backend自行维护一个定时器 控制反转 之前写代码的时候,有考虑一些控制反转,但是还需要细致下去,主要是为了做单元测试 不然现在改几行根本不敢确定有没有影响之前的功能 (未完待续)

By chainhelen

go

nsq 支持分布式事务(三)

主备同步 代码提交记录 完成度 只完成了主从复制;但是主备切换、依据处理能力动态调整消费能力、一些write失败逻辑处理 没有完成 基本逻辑 数据 主要分成了三部分数据的复制 1. 发送SLAVE_SYNC_INFO命令,返回所有的topic和其对应的channel 2. 发送SLAVE topic_name命令,携带当前topic的关键4个参数fileindex filenum totalMsgCnt viroffset和一个maxnum希望本次同步返回的消息数量,返回 master的关键参数和消息本身 3. 发送SLAVE topic_name channel_name,携带当前的4个关键参数,返回master的关键参数 客户端请求 对于slave是不可以接受普通client的请求的 ,连接master复用了master与client的 tcp端口 同步 or 异步 一开始写了一个同步版本,就是slave每次发送SLAVE消息,就等待结果 后来改成了异步发送,异步里面主要需要做大量的check避免时序问题,造成SYNC数据错乱问题

By chainhelen

nsq

nsq 支持分布式事务(二)

核心代码逻辑 commit 修改dtpre队列 在考虑落盘之前,先思考另一个问题,由于dtpre本身跟消费者无关,只跟生产者有关,其实并不需要等到消息从topic分发到各个channel里之后再进行commit(cacel),浪费存储空间(如果落盘就浪费了磁盘空间) 所以我们将dtpre的部分移至topic中存储 落盘逻辑 对于落盘逻辑来说,不只是dtpre的逻辑需要修改,如此对于一般的消息来说,都应该进行落盘 而channel来说只需要落盘数据索引即可 修改细节 实际这次提交修改了比较多的内容,最核心的还是移除了memoryMsgChan(无论是 topic 还是 channel),添加了backendMsgChan. 读写轮盘在于 1. nsqd/diskqueue_reader.go 采用了定时器pull方式,而不是之前的push方式 这个地方可能存在性能问题,因为每次是拉去tryReadOne的方式 2. nsqd/diskqueue_writer.go 里面writeOne函数,其实没有写入磁盘,而是写入了一个buffer(注意对于roc

By chainhelen

消息队列

go-nsq bug

起因 公司同事发现,其topic数据均摊在k8s里面 5个 nsqd 但是其中两个nsqd对应的消费channel已经出现积压,甚至落盘过万条数据 触发条件 注意每个 consumer 与 每个 NSQD 都有连接(图中没有画出) 每个consumer的maxinflight设置都是1 使用最新的官方sdk版本go-nsq v1.0.7 参数解释 这里面有两个参数需要解释一下,但是是从consumer的角度解释的,同样的参数在nsqd中可能含义不一样 RDY是代表当前消费者还可以接受多少条消息,进行处理(在上面场景,大部分情况都是0、1) maxinflight代表是当前还未完全处理完的消息最大条数(默认是1条) 负载均衡 例如从consumer A角度来看,有这样的场景,len(A.conns) > maxinflight 也就是说,consumer A在同一时间内,不能够RDY多个nsqd,所以只能暂时选取其中一个,然后过一段时间再来处理其他的nsqd上面的消息,实际在consumer里面确实有一个定时器去做这件事 fun

By chainhelen
nsq 支持分布式事务

go

nsq 支持分布式事务

原意 对于微服务架构来说,当数据垂直拆分到各个服务之后,通常通过调用接口的方式(或者异步消息)来保证数据一致性,但是如果出现被调用服务暂时性宕机(或者非平滑发布),那么两边数据就不一致了。 ps : 我认为还有一种情况,由于互联网人员流动性比较高,许多非核心业务中转好几个产品经理手,最后在新业务与旧业务联动的时候,会有遗漏一些业务异常情况,我暂时称之为"数据联动性",目前我除了规范业务流程,没有更好的方案;而本文所要解决的场景,称之为"分布式事务一致性" dt协议 对于A、B两个服务之间某个请求有分布式事务关联,使用如下方式,将分布式事务转化成单表事务 其中 send preapred 和 A服务的单表sql 顺序是可以随意的 实现这里需要修改一下nsq的代码,每一个client连接到nsqd都需要做一个协商(IDENTIFY)协议的版本号,通过magic字段来协商,目前nsq支持的是 V1和V2两个版本 所以我们需要添加一个文件protocol_dt.go,里面的内容主要是照搬protocol_v2.go,

By chainhelen