范塔西彬 - 技术分享
type: channel
link: https://fantasticbin.com/category/programming/
items: ElasticSearch 分页终极方案：search_after + Point-in-Time 原理与实践
items: 从策略模式到自引用泛型：Go 1.26 查询构建器的架构进化之路
items: Go语言构建用户行为分析系统：优雅架构设计与实战
items: 责任链模式：解耦与灵活流程处理的利器
items: Go泛型最佳实践：封装可扩展数据查询构建器
items: 无锁队列与延时无锁队列：原理、实现与应用
items: 软件测试的重要性：Go语言中的单元测试与Mock实践
items: 本地搭建ElasticSearch搜索引擎并同步MySQL数据实战
items: PHP、Go与MySQL中JSON数据处理与索引实践
items: Go并发编排之扇入模式

ElasticSearch 分页终极方案：search_after + Point-in-Time 原理与实践
type: item
date: 2026-05-18T11:30:00+08:00
description: 用 search_after 做深度分页的同学，有没有遇到过翻页数据重复或莫名丢失的情况？如果你的索引 refresh 频率比较高，这个坑大概率踩过。本文聊聊这个问题的根因，以及如何用 search_after + Point-in-Time (PIT) 彻底解决它。最后会结合 QueryBuilder 的实际代码，把整条链路串一遍。一、ElasticSearch 的分页困境1.1 from + size 的老问题ES 最基础的分页就是 from + size：GET /my_index/_search { "from": 10000, "size": 20, "query": { "match_all": {} }, "sort": [{ "created_at": "desc" }] }问题大家都清楚——翻得越深越慢。from = 10000, size = 20 时，每个分片要先取出 10020 条文档发给协调节点，协调节点再做全局排序、丢掉前 10000 条。这个开销随 from 线性增长，所以 ES 默认把 from + size 上限卡在 10000（index.max_result_window），超了直接报错。1.2 search_after：游标式深度分页search_after 是官方推荐的替代方案，思路很简单：不用偏移量了，拿上一页最后一条的排序值当游标，直接跳到下一页的起点。// 第一页 GET /my_index/_search { "size": 20, "query": { "match_all": {} }, "sort": [ { "created_at": "desc" }, { "_id": "asc" } ] } // 第二页：把第一页最后一条的 sort values 塞进来 GET /my_index/_search { "size": 20, "query": { "match_all": {} }, "sort": [ { "created_at": "desc" }, { "_id": "asc" } ], "search_after": ["2026-05-17T10:30:00Z", "doc_999"] }好处很明显：不管翻到第几页，每次查询开销都是固定的（只取 size 条）不受 max_result_window 限制但这里有个隐含前提——search_after 假设两次请求之间，索引的数据和排序是稳定的。一旦这个前提被打破，问题就来了。二、高频 refresh 下的排序不稳定问题2.1 问题根因ES 是近实时（Near Real-Time）搜索引擎，写入的文档不会立即可搜索，要等 refresh 之后才对搜索可见。默认 refresh 间隔是 1 秒（index.refresh_interval）。这就带来一个问题：用 search_after 做跨请求分页时，两次请求之间很可能发生 refresh，导致：新文档被 refresh 进来——如果它的排序值恰好落在已翻过的页和当前页之间，后续页的文档整体"后移"，你会看到重复数据旧文档被删除或更新——排序值变了，某些文档直接被跳过段合并（Segment Merge）——后台合并改变了文档物理存储顺序，影响 _doc 等隐式排序画个时间线就很清楚了：请求1（第1页） 请求2（第2页） │ │ ▼ ▼ ┌─────────┐ refresh 发生 ┌─────────┐ │ 文档A │ ──────────► │ 文档X(新)│ ← 新文档插入 │ 文档B │ │ 文档A │ ← 文档A 重复出现！ │ 文档C │ │ 文档B │ │ ... │ │ ... │ └─────────┘ └─────────┘2.2 影响场景这个问题在以下场景里比较明显：场景影响程度说明数据导出🔴 严重全量遍历时数据重复或遗漏，导出结果不完整后台管理列表翻页🟡 中等用户翻页看到重复数据，体验差移动端"加载更多"🟡 中等下拉加载出现已看过的内容实时数据流消费🔴 严重消费位点不准确，可能重复处理或丢失数据2.3 为什么加 _id 排序也没用？可能有人会想：加一个唯一的 _id 作为 tiebreaker 排序字段不就行了？_id 确实能保证排序的唯一性，但它没法阻止新文档插入到已翻过的区间。search_after 的语义是"给我排序值大于这个游标的文档"——refresh 后如果出现了新的符合条件的文档，它们照样会出现在结果里。所以根本问题不在排序是否唯一，而在于两次查询看到的数据视图不一致。三、Point-in-Time (PIT)：给索引拍个快照3.1 PIT 是什么？Point-in-Time (PIT) 是 ES 7.10 引入的特…

从策略模式到自引用泛型：Go 1.26 查询构建器的架构进化之路
type: item
date: 2026-04-20T14:06:00+08:00
description: 当泛型遇上自引用约束，代码的表达力将迎来质的飞跃Go 语言的泛型自 1.18 版本引入以来，一直存在一个令人遗憾的限制——泛型类型不能在自身的类型参数列表中引用自己。这意味着诸如"链式调用返回具体子类型"这样在 Java/C# 中司空见惯的模式，在 Go 中只能通过 any 类型断言来曲线救国。直到 Go 1.26 的发布，这一限制终于被打破。本文将继续对《Go泛型最佳实践：封装可扩展数据查询构建器》中介绍的查询构建器进行版本迭代，带你深入理解这一特性如何彻底改变架构设计。Go 1.26 自引用泛型约束：到底解锁了什么？Go 1.26 版本正式解除了"泛型类型不得在自身类型参数列表中自我引用"的限制。简单来说，以前你不能这样写：// Go 1.25 及之前：编译错误！ type Builder[B Builder[B]] struct { // ... }而在 Go 1.26 中，这段代码完全合法。这意味着我们可以定义这样的接口约束：type queryBuilder[B any, R any] interface { self() B QueryList(ctx context.Context) ([]*R, int64, error) } // Go 1.26：合法！ type builder[B queryBuilder[B, R], R any] struct { data *DBProxy start uint32 limit uint32 needTotal bool needPagination bool middlewares []Middleware[R] selfRef B // 存储具体子类型引用 querierRef Querier[R] // 存储 Querier 接口引用 }这里的关键在于 B 类型参数——它代表"实现者自身的类型"。当一个具体的构建器（如 GormBuilder）实现这个接口时，B 就是 *GormBuilder[R]，这样 self() 方法就能返回具体的子类型，而非笼统的接口类型。这解决了什么问题？ 一句话概括：让基类的方法能够返回子类的具体类型，实现真正类型安全的链式调用。痛点回顾：旧架构的"类型断言之殇"在 Go 1.26 之前，我们的查询构建器采用经典的策略模式。核心思路是：一个通用的 builder 结构体持有一个 Strategy 接口，不同数据源（MySQL、MongoDB、Elasticsearch）各自实现该接口。旧版核心结构// 旧版：单一构建器 + 策略接口 type builder[R any] struct { data *DBProxy start uint32 limit uint32 needTotal bool needPagination bool strategy Strategy[R] // 查询策略 middlewares []Middleware[R] // 中间件链 filter func(context.Context) (any, error) // 返回 any！ sort func() any // 返回 any！ } // 策略接口 type Strategy[R any] interface { QueryList(context.Context, *builder[R]) ([]*R, int64, error) }旧版策略实现（以 GORM 为例）func (s *QueryGormListStrategy[R]) QueryList( ctx context.Context, builder *builder[R], ) (list []*R, total int64, err error) { filterScope, err := builder.filter(ctx) if err != nil { return nil, 0, err } sortScope := builder.sort() // 😱 运行时类型断言！如果类型不匹配，直接 panic for _, scope := range []any{filterScope, sortScope} { if _, ok := scope.(func(*gorm.DB) *gorm.DB); !ok { return nil, 0, errors.New("invalid scope") } } // 使用时还需要再次断言 query := builder.data.db.WithContext(ctx). Model(&list). Scopes(filterScope.(func(*gorm.DB) *gorm.DB), sortScope.(fun…

Go语言构建用户行为分析系统：优雅架构设计与实战
type: item
date: 2025-12-24T23:29:00+08:00
description: 构建高性能用户行为分析系统：Go语言实现优雅架构设计用户行为分析，是产品迭代的"透视镜"在互联网产品"唯快不破"的今天，无数团队陷入同一个怪圈：埋头开发功能、满怀期待发布，最终却对用户的真实反馈一无所知。本文将带你从后端视角，拆解如何构建一套既稳健又灵活的用户行为分析处理流程。为什么要深度投入用户行为分析？从硅谷科技巨头到国内独角兽企业，再到每个追求精细化运营的创业团队，用户行为分析早已不再是锦上添花的"奢侈品"，而是决定产品生死的"标配能力"。行业数据清晰表明：采用深度用户行为分析的产品，其迭代效率与商业转化率通常能实现数倍提升。我们不再依赖主观猜测，而是让数据说话——每一次点击、每一次滑动、每一秒的停留，都在无声地告诉我们：用户是谁？他们真正想要什么？又是在哪个环节默默离开？今天，我们就从后端架构出发，探讨如何优雅地打造这套"产品听诊器"。架构全景：一个优雅的幕后指挥官假设前端已预埋用户事件，并通过消息队列（如Kafka、RabbitMQ等）投递事件消息。后端的核心挑战在于：如何高效、稳健地处理这些如洪峰般涌来的数据？我们需要一个强大的"指挥官"来调度全局，我们将其命名为 UserBehaviorAnalyzer。它必须具备四大核心能力：流量控制：防止瞬时流量冲垮数据库数据清洗：将原始JSON转化为业务可用的结构化数据策略分发：根据业务需求（如点击热力图、用户路径分析、转化率统计）分发至不同处理模块并发执行：利用Go协程优势并行处理，提升吞吐量下面让我们从零开始，一步步构建这个系统。1. 策略规范接口：定义通用语言首先，我们定义策略处理的通用接口，这是整个系统的"契约"：package mq import ( "context" "fmt" ) // StrategyContext 定义策略处理数据的通用接口类型 // 具体策略在实现时，需要将此类型断言为具体的策略上下文结构 type StrategyContext any // ProcessingStrategy Kafka上报消息处理策略接口 type ProcessingStrategy interface { // GetName 返回该策略处理的name值 GetName() string // Process 处理具体的业务逻辑 Process(ctx context.Context, data StrategyContext) error // IsApplicable 判断该策略是否适用于当前上下文 IsApplicable(ctx context.Context, name string, data StrategyContext) bool }2. 策略注册表：集中管理所有"处理器"接下来实现策略的注册与发现机制，通过读写锁保护策略字典，确保并发安全：package mq import ( "sync" ) // StrategyRegistry 策略注册表 type StrategyRegistry struct { strategies map[string]ProcessingStrategy mu sync.RWMutex } // NewStrategyRegistry 创建新的策略注册表 func NewStrategyRegistry() *StrategyRegistry { return &StrategyRegistry{ strategies: make(map[string]ProcessingStrategy), } } // Register 注册策略 func (r *StrategyRegistry) Register(strategy ProcessingStrategy) { r.mu.Lock() defer r.mu.Unlock() r.strategies[strategy.GetName()] = strategy } // GetStrategy 获取指定name对应的策略 func (r *StrategyRegistry) GetStrategy(name string) ProcessingStrategy { r.mu.RLock() defer r.mu.RUnlock() if strategy, exists := r.strategies[name]; exists { return strategy } return nil } // GetAllStrategies 获取所有策略 func (r *StrategyRegistry) GetAllStrategies() []ProcessingStrategy { r.mu.RLock() defer r.mu.RUnlock() …

责任链模式：解耦与灵活流程处理的利器
type: item
date: 2025-07-05T18:41:00+08:00
description: 在日常工作中，我们常常会遇到产品增加需求的情况，或者需要设计一个工作流引擎。如果采用硬编码的方式，不仅不利于代码的解耦，后期的维护工作也会让人头疼不已。这时，责任链（Chain of responsibility）模式就能派上用场。责任链模式概述责任链模式是一种行为型设计模式。通过使用该模式，我们可以为请求创建一条由多个处理器组成的链路。每个处理器专注于自己的职责，彼此之间不存在耦合关系。当一个处理器完成自己的任务后，请求对象会被传递到链路中的下一个处理器继续处理。责任链模式在许多流行框架中都有广泛应用，例如中间件和拦截器等框架组件就采用了这种设计模式。在进行 Web 接口开发时，像记录访问日志、解析 Token、格式化接口响应的统一结构等项目公共逻辑，通常都在中间件和拦截器中完成。这样做可以将这些基础操作与接口的业务逻辑解耦。责任链模式的实现示例：医院患者就诊流程PHP 实现责任链模式使用面向对象语言实现非常简单，因为可以通过抽象类复用公共逻辑，链路的执行逻辑也可以由各个处理器自行实现。下面我们以医院患者就诊流程为例，展示如何使用 PHP 实现责任链模式。declare(strict_types=1); // 就诊状态枚举 enum PatientStatus: int { case Start = 1; // 状态：开始看病 case Reception = 2; // 状态：已挂号 case Clinic = 3; // 状态：已看诊 case Cashier = 4; // 状态：已缴费 case Pharmacy = 5; // 状态：已拿药 } // 患者类 class Patient { public ?PatientStatus $status = null; public function __construct( public string $name ) {} } // 就诊流程责任链接口定义 interface PatientHandler { public function execute(Patient $patient): void; public function setNext(PatientHandler $handler): PatientHandler; public function do(Patient $patient): void; } // 抽象类复用公共定义 abstract class BaseHandler implements PatientHandler { private ?PatientHandler $nextHandler = null; public function execute(Patient $patient): void { $this->do($patient); $this->nextHandler?->execute($patient); } public function setNext(PatientHandler $handler): PatientHandler { $this->nextHandler = $handler; return $handler; } abstract public function do(Patient $patient): void; } class Reception extends BaseHandler { public function do(Patient $patient): void { if ($patient->status > PatientStatus::Start) { echo "患者已经挂号\n"; return; } echo "挂号处为患者 {$patient->name} 挂号\n"; $patient->status = PatientStatus::Reception; } } class Clinic extends BaseHandler { public function do(Patient $patient): void { if ($patient->status === PatientStatus::Start) { echo "患者还未挂号\n"; return; } if ($patient->status > PatientStatus::Reception) { echo "患者已经完成医生诊断\n"; return; } echo "医生为患者 {$patient->name} 进行诊断\n"; $patient->status = PatientStatus::Clinic; } } class Cashier extends BaseH…

Go泛型最佳实践：封装可扩展数据查询构建器
type: item
date: 2025-04-20T15:11:00+08:00
description: 日常迭代管理后台时，有一个经常需要维护的需求，那就是查询数据列表，包含的一系列功能包括但不限于分页、筛选、排序、导出、批量操作等。这些功能虽然都是大同小异，但却需要在不同的场景下进行不同的配置，这就导致了代码的冗余和维护成本的增加。最近在工作中发现 Go 语言的泛型功能可以很好地解决这个问题，通过泛型可以将一些通用的功能抽象出来，达到代码复用的目的。每次不同的数据场景只需要进行泛型推断，然后定义各自的查询细节，就可以实现不同场景下的查询功能，不需要重复编写相同的代码。下面我们就来实现一个通用的列表查询构建器，不同的数据源通过策略模式进行注入，然后通过泛型变量来实现相同场景下的数据定义。话不多说，我们直接看代码。假设我们现在参与了一个商城项目，在管理后台中需要查询商品列表，包括商品名称、价格、库存等信息。我们可以定义一个商品实体：package model import "gorm.io/gorm" type Goods struct { Name string `gorm:"column:name;type:varchar(100);not null;comment:商品名称"` Price float32 `gorm:"column:price;type:decimal unsigned;not null;comment:商品价格"` Stock uint32 `gorm:"column:stock;type:int unsigned;not null;comment:商品库存"` Status uint32 `gorm:"column:status;type:tinyint unsigned;not null;default:1;comment:商品状态"` gorm.Model } func (*Goods) TableName() string { return "goods" }再将 goods.proto 文件定义如下：syntax = "proto3"; package api.goods.v1; import "google/api/annotations.proto"; option go_package = "kratos-demo/api/goods/v1;v1"; option java_multiple_files = true; option java_package = "api.goods.v1"; service Goods { rpc ListGoods (ListGoodsRequest) returns (ListGoodsReply) { option (google.api.http) = { get: "/v1/goods" }; }; } message ListGoodsRequest { ListGoodsFilter filter = 1; QueryGoodsListSort sort = 2; uint32 start = 3; uint32 limit = 4; } message ListGoodsReply { repeated GoodsInfo goods = 1; uint32 total = 2; } message ListGoodsFilter { string name = 1; float min_price = 2; float max_price = 3; } enum QueryGoodsListSort { CREATED_AT = 0; PRICE = 1; } enum GoodsStatus { NONE = 0; NORMAL = 1; OFFLINE = 2; } message GoodsInfo { uint64 id = 1; string name = 2; float price = 3; uint32 stock = 4; GoodsStatus status = 5; string created_at = 6; }开始进行准备工作，我们使用 errgroup 来定义一个并发查询列表及总条数的工具方法：package utils import ( "fmt" "golang.org/x/sync/errgroup" "runtime/debug" ) // WaitAndGo 等待所有函数执行完毕 func WaitAndGo(fn ...func() error) error { defer func() { if err := recover(); err != nil { fmt.Printf("Panic: %+v\n %s", err, string(debug.Stack())) } }() var g …

无锁队列与延时无锁队列：原理、实现与应用
type: item
date: 2024-12-03T16:29:00+08:00
description: 在之前的文章使用atomic包实现无锁并发统计中，我们了解到原子操作在保证并发安全的同时，能避免锁带来的性能损耗，其应用场景广泛，如并发计数器、自旋锁等。这次，我们将深入探讨原子操作的另一个经典应用场景 —— lock-free queue（无锁队列），并基于此实现一个延时无锁队列。队列基础队列是一种遵循先进先出（FIFO，First In First Out）原则的数据结构，最先进入队列的元素最先被移除，元素顺序得以保持。队列结构包含队首和队尾，支持入队和出队操作。常见的队列实现方式有数组和链表。数组实现的队列在入队和出队时可能需要移动元素，而链表实现的队列则因只需更新指针，操作更为高效。无锁队列实现无锁队列利用原子操作，在保证并发安全的前提下，允许多个线程同时进行入队和出队操作，无需传统锁机制，显著提升并发性能。我们采用链表结构来进一步优化效率，其定义如下：import ( "sync/atomic" "unsafe" ) type LkQueue[T any] struct { head unsafe.Pointer tail unsafe.Pointer } type node[T any] struct { value T next unsafe.Pointer }基础操作实现首先，我们完善队列的创建、读取节点及修改节点的操作：func NewLkQueue[T any]() *LkQueue[T] { n := unsafe.Pointer(&node[T]{}) return &LkQueue[T]{head: n, tail: n} } // load读取节点的值 func load[T any](p *unsafe.Pointer) *node[T] { return (*node[T])(atomic.LoadPointer(p)) } // cas原子地修改节点的值 func cas[T any](p *unsafe.Pointer, old, new *node[T]) bool { return atomic.CompareAndSwapPointer(p, unsafe.Pointer(old), unsafe.Pointer(new)) }这里的读取节点（load）和修改节点（cas）操作使用了原子读及 CAS 操作，确保在并发访问时数据安全。入队操作入队操作流程如下：先接收值并创建新节点，然后循环尝试将新节点插入队列尾部。每次循环中，通过 load 函数读取当前队列的尾部节点 tail 和其下一个节点 next，检查 tail 是否仍为队列尾部（确保状态未变）。若 tail 是队列尾部，再检查 next 是否为 nil。若 next 为 nil，尝试将新节点 n 插入 tail 的 next 位置；若插入成功，进一步更新队列的 tail 指针指向新节点 n（均通过原子操作 CAS 完成）；若插入失败，说明队列状态被其他线程修改，继续循环尝试插入。若 next 不为 nil，说明 tail 不是队列最后一个节点，尝试将 tail 指针移动到 next 节点，以便下次循环继续尝试插入。// Enqueue入队 func (q *LkQueue[T]) Enqueue(value T) { n := &node[T]{value: value} for { tail := load[T](&q.tail) next := load[T](&tail.next) if tail == load[T](&q.tail) { // tail和next是否一致 if next == nil { if cas(&tail.next, next, n) { cas(&q.tail, tail, n) // 入队完成。设置tail return } } else { cas(&q.tail, tail, next) } } } }出队操作出队操作同样在操作前进行检查。代码循环尝试从队列头部移除节点。每次循环中，通过 load 函数读取当前队列的头部节点 head、尾部节点 tail 和头部节点的下一个节点 next，检查 head 是否仍为队列头部（确保状态未变）。若 head 仍为队列头部，进一步检查 head 和 tail 是否相等。若 head 和 tail 相等，队列可能为空或 tail 未到队尾，此时检查 next 是否为 nil ；若 next 为 nil，说明队列为空，返回 false 表示出队失败；若 next 不为 nil，说明 tail 未到队尾，尝试将 tail 指针移动到 next 节点，以便下次循环继续出队操作。若 head 和 tail 不相等，说明队列中有元素可出队，将 next 节点…

软件测试的重要性：Go语言中的单元测试与Mock实践
type: item
date: 2024-05-26T10:40:00+08:00
description: 是的，我又离职了。这一次离开的主要原因是与上家公司团队理念上的不合，可能还包括一些其他因素。真正让我决定离开的核心原因是这个团队乃至整个公司都不注重测试。这或许与公司产品定位有关，因为他们是一家游戏公司，需要快速抓住用户市场并推出爆品来盈利，导致产品周期迭代过快，根本没有多余时间进行测试。此外，缺乏一个好的发布流程和 Review 机制，从而导致生产事故频发。测试的重要性一般软件开发公司会专门设立 QA（Quality Assurance）工程师职位，专门负责把控产品质量。部分职责包括从业务角度（特别是边缘场景）进行测试，确保用户在使用产品过程中不会遇到问题及 bug，然后及时反馈并参与到迭代周期中，这是产品质量控制中不可或缺的一环。单元测试的重要性从开发工程师的角度把控产品质量的方式之一就是单元测试。开发者写的每个应用程序、业务流程、功能模块甚至每个独立的小单元，都必须通过单元测试确定是否达到预期。一些规范化的敏捷开发团队会将单元测试纳入 CI/CD 的持续集成中，实现了自动化构建部署一套体系。而在单元测试中，Mock 对象的使用非常普遍，尤其在需要隔离被测试代码与外部依赖（如数据库、网络服务、文件系统等）时。Mock 对象可以模拟这些外部依赖的行为，从而使测试更加独立和可控。今天我们就来了解一下 Go 语言单元测试中对于 Mock 所做的支持。使用 Gomock 进行单元测试Go 语言官方提供了一款 Mock 框架：Gomock 以及 Mock 文件生成工具：mockgen。它可以方便地生成模拟接口，支持与Go的内置 testing 包整合，也可以在其他上下文中使用。该项目源自 Google 的 golang/mock 仓库，但 Google 已不再维护。鉴于 Uber 内部对 Gomock 的大量使用，Uber 决定接手并继续维护该项目。我们将基于 uber-go/mock 做示例。使用步骤Gomock 的使用通常遵循以下四个基本步骤：生成 Mock 文件：使用 mockgen 为你想要 mock 的接口生成一个 mock 文件。创建 Mock 对象：在测试代码中，创建一个 gomock.Controller 实例，并将其作为参数传递给 mock 对象的构造函数来创建 mock 对象。设置期望：调用 EXPECT() 为 mock 对象设置期望和返回值。验证期望：调用 mock 控制器的 Finish() 方法验证 mock 的期望行为（通常在 defer 中完成）。示例通过一个简单的例子演示 Gomock 的使用流程。假设你有一个简单的服务接口 UserService，它有一个方法 GetUser，通过用户ID获取用户信息。我们将为这个接口创建一个 mock 对象并编写测试代码。首先，在项目中安装 mockgen，并加入到环境变量中：go install go.uber.org/mock/mockgen@latest # 配置环境变量 ... go get -u go.uber.org/mock/gomock go get -u github.com/stretchr/testify/assert接着，在 service/user.go 中定义一个接口，里面有一个 GetUser() string 方法，模拟获取信息。package service type User struct { ID int Name string } type UserService interface { GetUser(id int) (*User, error) }使用 Mockgen 生成 UserService 的 mock 对象：mockgen -source=service/user.go -destination=service/mock_user.go -package=service这里的 destination 参数通常设置成与 source 参数文件同级目录，否则运行 go test 会出现 import cycle not allowed in test，也就是循环导入错误；package 包名一般跟目录名相同即可。假设我们有一个函数 GetUserName，它使用 UserService 接口实例中的方法来获取用户并返回用户的名字：package service func GetUserName(us UserService, id int) (string, error) { // 使用依赖注入进来的具体结构体方法，通常为需要 mock 的方法 user, err := us.GetUser(id) if err != nil { return "", err } return user.Name, nil }在…

本地搭建ElasticSearch搜索引擎并同步MySQL数据实战
type: item
date: 2024-05-01T08:54:00+08:00
description: 这次我们将搭建一个 ElasticSearch 搜索引擎，并使用阿里开源的 Canal 中间件实现 MySQL 中用户数据的实时同步。我们知道 MySQL 的主备复制流程如下：master 将数据变更写入二进制日志（binary log），记录为二进制日志事件（binary log events），可通过 show binlog events 查看。slave 将master 的二进制日志事件复制到中继日志（relay log）。slave 重放中继日志中的事件，将数据变更反映到自己的数据中。Canal [kə’næl]，译为水道/管道/沟渠，主要用于基于 MySQL 数据库的增量日志解析，提供增量数据订阅和消费，不支持全量数据同步。由于它采用了 Binlog 机制，因此 MySQL 的新增、更新、删除操作对应的 ElasticSearch 都能实时响应。Canal 主要实现以下几项任务：模拟 MySQL slave 的交互协议，伪装为 MySQL slave ，向 MySQL master 发送 dump 协议。MySQL master 收到 dump 请求后，开始向 slave （即 Canal）推送 binary log。Canal 解析 binary log 对象（原始为 byte 流，可转换为 protocol 进行处理）。接下来，我们通过一个 docker-compose.yaml 文件来搭建 MySQL 8.0.36、Elasticsearch 7.17.20 和 Canal 1.1.7。version: '3.8' services: mysql: container_name: canal-demo-mysql image: mysql:8.0.36 restart: always environment: MYSQL_ROOT_PASSWORD: 123456 ports: - "3306:3306" networks: - canal-demo elasticsearch: container_name: canal-demo-elasticsearch image: elasticsearch:7.17.20 restart: always environment: - "discovery.type=single-node" - "ES_JAVA_OPTS=-Xms512m -Xmx512m" ports: - "9200:9200" - "9300:9300" networks: - canal-demo canal: container_name: canal-demo-canal image: canal/canal-server:v1.1.7 restart: always ports: - "11111:11111" environment: - "CANAL_ADMIN_MANAGER=admin" - "CANAL_ADMIN_PORT=11111" - "CANAL_USER=admin" - "CANAL_PASS=admin" networks: - canal-demo networks: canal-demo: driver: bridge执行 docker-compose up -d 启动容器编排。接下来进入 MySQL 容器，创建数据库和用户表，并给 name 字段添加索引以提高查询效率。docker exec -it mysql容器ID mysql -uroot -p # 输入密码 123456 进入 MySQL 命令行CREATE DATABASE mydb; USE mydb; CREATE TABLE `users` ( `id` bigint unsigned NOT NULL AUTO_INCREMENT, `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `birthday` date NOT NULL, `created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`), INDEX `idx_name` (`name`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;退出容器后，为 ElasticSearch 创建一个同步…

PHP、Go与MySQL中JSON数据处理与索引实践
type: item
date: 2024-04-06T15:34:00+08:00
description: PHP 在 8.3 的版本中增加了 json_validate() 函数可用于检查一个字符串是否为语法正确的 JSON，同时官方号称比 json_decode() 更有效。在 8.3 版本之前，如果需要实现 JSON 验证，则需要使用 json_decode() 来辅助实现，并且通过验证最近是否存在 JSON_ERROR_NONE 错误来达到效果：// PHP < 8.3 function json_validate(string $string): bool { json_decode($string); return json_last_error() === JSON_ERROR_NONE; } var_dump(json_validate('{ "test": { "foo": "bar" } }')); // true而在 8.3 之后，只需要通过 json_validate() 函数就能直接验证字符串了：var_dump(json_validate('{ "test": { "foo": "bar" } }')); // true如果在 Go 语言中需要用到 JSON 验证，官方 encoding/json 包中已经定义了 json.Valid 来实现：import ( "encoding/json" "fmt" ) func main() { fmt.Println(json.Valid([]byte(`{ "test": { "foo": "bar" } }`))) // true }同时我们知道 MySQL 5.7 及更高版本中可以为 JSON 字段的某个 key 设置索引来高效检索，可支持两种 Generated Column，即 Virtual Generated Column 和 Stored Generated Column，前者只将 Generated Column 保存在数据字典中（表的元数据），并不会将这一列数据持久化到磁盘上；后者会将 Generated Column 持久化到磁盘上，而不是每次读取的时候计算所得。很明显，后者存放了可以通过已有数据计算而得的数据，需要更多的磁盘空间，与 Virtual Column 相比并没有优势，因此，MySQL 5.7 中，不指定 Generated Column 的类型，默认是 Virtual Column。--- 表结构 CREATE TABLE `users` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `name` VARCHAR(255) NOT NULL DEFAULT '', `information` JSON, `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP, `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `deleted_at` DATETIME DEFAULT NULL ); --- 创建JSON字段的虚拟列，user 是包含 JSON 字段的表名，information 是 JSON 字段的列名，label 是要索引的键名，label_index 是新添加的虚拟列名。 ALTER TABLE users ADD COLUMN label_index VARCHAR(255) AS (information->>'$.label') STORED; --- 为虚拟列创建索引 CREATE INDEX idx_label_index ON users(label_index);然后可以使用市面上成熟的 ORM 来便捷操作 JSON 字段，在 Laravel 中我们可以先使用 make:model Artisan 命令来生成模型类：php artisan make:model User我们再来看看在 Eloquent ORM 中使用 JSON 字段的示例：// 新增 $user = new User; $user->name = 'fantasticbin'; $user->information = ['label' => 'cool', 'language' => ['php', 'go']]; $user->save(); // 修改 $user = User::find(1); $information = $user->information; $information['label'] = 'nice'; $user->information = $information; $user->save(); …

Go并发编排之扇入模式
type: item
date: 2024-03-04T19:11:00+08:00
description: 在以往的博文中我们知道 Go 可以通过 chan 可以实现很多种方式的并发编排，包括 Or-Done 模式、扇入模式、扇出模式、Stream 和 map-reduce 等，并且已经基于扇出模式实现了设计模式之观察者模式，这次我们就来探讨一下扇入模式。扇入模式的定义跟扇出模式是反着来的，即通过多个输入 chan 将结果合并到一个输出 chan 中并返回。举个实际的栗子，假设我们有两个上游，需要将上游数据合并成一个统一的下游进行输出，代码定义如下：func mergeTwo[T any](ch1, ch2 <-chan T) <-chan T { out := make(chan T) go func() { defer close(out) for ch1 != nil || ch2 != nil { // 至少有一个 chan 不是 nil 则继续 select { case v, status := <-ch1: if !status { ch1 = nil // ch1 关闭时，将其赋值为 nil break } out <- v case v, status := <-ch2: if !status { ch2 = nil // ch2 关闭时，将其赋值为 nil break } out <- v } } }() return out }这里有一个 select 动态监听的小技巧，也就是将监听的 chan 设置为 nil 后，就不会再走入到该分支中了，通过使用这种方式可以做到 select 动态去除某个 chan 的监听动作。如果是存在多个上游的情况，那么如何实现扇入模式的合并逻辑？一种方式是通过递归来实现，思路就是将输入 chan 进行两两合并：func mergeRecursion[T any](chs ...<-chan T) <-chan T { switch len(chs) { case 0: // 输入 chan 的数量为0，返回一个已关闭的 chan out := make(chan T) close(out) return out case 1: // 输入 chan 的数量为1，直接返回 return chs[0] case 2: // 输入 chan 的数量为2，合并这两个 chan return mergeTwo(chs[0], chs[1]) default: m := len(chs) / 2 return mergeTwo( mergeRecursion(chs[:m]...), // 递归调用前半部分 mergeRecursion(chs[m:]...)) // 递归调用后半部分 } }这里使用前面的 mergeTwo 方法来处理两个输入 chan 的合并，最终可将所有的输入 chan 进行合并输出。另一种方式则可以用到 reflect.SelectCase 来处理输入 chan 的接收，把结果直接丢入输出 chan 中即可。import "reflect" func fanIn[T any](chs ...<-chan T) <-chan T { out := make(chan T) go func() { defer close(out) var cases []reflect.SelectCase for _, ch := range chs { cases = append(cases, reflect.SelectCase{ Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch), }) } for len(cases) > 0 { i, value, ok := reflect.Select(cases) // 从输入 chan 中读取数据 if !ok { cases = append(cases[:i], cases[i+1:]...) // 将已关闭的输入 chan 从 select 列表中移除 continue } out <- value.Interface().(T) // 这里需要从 interface 中转换成具体的类型 } }() return out }通过动态去除 reflect.SelectCase 切片中的元素也能解决 chan 动态监听的场景，同时能够优雅处理多个输入上游的监听逻辑。本文代码部分引用《 100个Go语言典型错误 》之 #66：没有使用 nil channel，及晁岳攀（@鸟窝）大佬的《 深入理解Go并发编程 》之第 12 章 - channel 的内部实现和陷阱。