【基于 SwooleHyperf 的微服务实战】第九周·周三打造商品服务和库存服务今天我们进入第九周周三继续构建电商核心系统。昨天用户服务已经上线今天我们将同时打造商品服务和库存服务。它们是电商的货架和仓库负责商品展示、库存扣减并为后续的订单和秒杀提供底层支撑。你将亲手创建这两个服务实现它们的 RPC 接口并通过 Consul 注册发现使网关能够聚合商品数据。今日目标创建product-service和inventory-service两个独立 Hyperf 项目。实现商品服务的GetProductById、ListProducts等接口连接product_db。实现库存服务的Freeze、Unfreeze、Deduct、GetStock接口连接inventory_db。在库存服务中集成Redis 分布式锁保护冻结/解冻的并发安全。配置网关路由通过聚合接口/products返回商品列表含库存信息。测试商品浏览和库存扣减流程验证服务间 RPC 调用和 Consul 发现。一、项目创建与基础配置约 45 分钟我们将同时创建两个服务过程与昨天用户服务类似但更精炼。1. 创建商品服务docker-composeexecswoolebashcd/var/www/servicescomposercreate-project hyperf/hyperf-skeleton productcdproductcomposerrequire hyperf/database hyperf/redis hyperf/json-rpc hyperf/service-governance-consul编辑.env设置数据库为product_db端口9503避免冲突。配置config/autoload/server.php添加 JSON-RPC 服务器监听9503servers[[namejsonrpc,typeServer::SERVER_HTTP,host0.0.0.0,port9503,callbacks[Event::ON_REQUEST[\Hyperf\JsonRpc\HttpServer::class,onRequest],],],],2. 创建库存服务cd/var/www/servicescomposercreate-project hyperf/hyperf-skeleton inventorycdinventorycomposerrequire hyperf/database hyperf/redis hyperf/json-rpc hyperf/service-governance-consul hyperf/redis-lock端口设为9504数据库inventory_db。配置server.php监听9504。3. 准备数据库表确保 MySQL 中已存在product_db和inventory_db并包含昨天设计的表。如果尚未创建可手动执行 SQLdatabases/目录下的文件应该已通过 Docker 卷自动加载。我们填充一些示例商品数据USEproduct_db;INSERTINTOproducts(name,price,stock)VALUES(Hyperf微服务实战,99.00,100),(Swoole协程指南,79.00,50);USEinventory_db;INSERTINTOstocks(product_id,total)VALUES(1,100),(2,50);注意stock表中的frozen初始为 0available自动计算二、知识核心服务间数据关联与库存锁约 1 小时1. 商品与库存的职责分离商品服务管理商品基本信息名称、价格库存服务管理实时库存。商品列表需要展示库存但两个服务数据库独立。我们有几种方案网关聚合网关分别调商品和库存服务合并数据后返回类似/feed聚合。商品服务缓存库存商品服务监听库存变更事件本地缓存库存量。实时 RPC每次查询商品列表时商品服务调用库存服务批量获取库存。今天我们采用网关聚合方案因为它最解耦且符合架构职责。2. 库存服务的并发安全Freeze冻结和Unfreeze解冻操作在高并发下必须保证数据一致。我们使用 Redis 分布式锁保护同一商品的库存操作防止超卖。锁键为lock:inventory:{productId}在服务内部使用hyperf/redis-lock注解或手动加锁。三、实战编写商品与库存服务接口约 2.5 小时步骤 1商品服务实现定义接口app/JsonRpc/ProductServiceInterface.php?phpnamespaceApp\JsonRpc;useHyperf\RpcServer\Annotation\RpcService;#[RpcService(name:ProductService,protocol:jsonrpc-http,server:jsonrpc)]interfaceProductServiceInterface{publicfunctiongetProductById(int$id):array;publicfunctionlistProducts(int$page,int$pageSize):array;publicfunctioncreateProduct(string$name,float$price,int$stock):array;}实现类app/JsonRpc/ProductService.php?phpnamespaceApp\JsonRpc;useApp\Model\Product;useHyperf\DbConnection\Db;#[RpcService(name:ProductService,protocol:jsonrpc-http,server:jsonrpc)]classProductServiceimplementsProductServiceInterface{publicfunctiongetProductById(int$id):array{$productProduct::find($id);return$product?$product-toArray():[];}publicfunctionlistProducts(int$page1,int$pageSize10):array{$paginatorProduct::paginate($pageSize,[*],page,$page);return[products$paginator-items(),total$paginator-total(),];}publicfunctioncreateProduct(string$name,float$price,int$stock):array{$productProduct::create([name$name,price$price,stock$stock,]);// 同步创建库存记录通过 RPC 调用库存服务或直接在同一事务中但库存服务独立所以这里仅记录库存需额外初始化// 为了简化我们在创建商品时通过本地数据库操作同时向 inventory_db 写入不能跨库。// 正确的做法由协调器或网关先后调用商品和库存服务。return$product-toArray();}}模型app/Model/Product.php?phpnamespaceApp\Model;useHyperf\DbConnection\Model\Model;classProductextendsModel{protected$tableproducts;protected$fillable[name,price,stock];}依赖绑定config/autoload/dependencies.phpApp\JsonRpc\ProductServiceInterface::classApp\JsonRpc\ProductService::class,步骤 2库存服务实现接口app/JsonRpc/InventoryServiceInterface.php?phpnamespaceApp\JsonRpc;useHyperf\RpcServer\Annotation\RpcService;#[RpcService(name:InventoryService,protocol:jsonrpc-http,server:jsonrpc)]interfaceInventoryServiceInterface{publicfunctionfreeze(int$productId,int$quantity):bool;publicfunctionunfreeze(int$productId,int$quantity):bool;publicfunctiondeduct(int$productId,int$quantity):bool;publicfunctiongetStock(int$productId):array;}实现类app/JsonRpc/InventoryService.php?phpnamespaceApp\JsonRpc;useHyperf\DbConnection\Db;useHyperf\Redis\Redis;useHyperf\Di\Annotation\Inject;#[RpcService(name:InventoryService,protocol:jsonrpc-http,server:jsonrpc)]classInventoryServiceimplementsInventoryServiceInterface{#[Inject]privateRedis$redis;publicfunctionfreeze(int$productId,int$quantity):bool{// 分布式锁保护同一商品的库存操作$lock$this-redis-lock(lock:inventory:.$productId,5);try{if(!$lock-get()){returnfalse;// 获取锁失败可重试}// 检查可用库存$stockDb::table(stocks)-where(product_id,$productId)-first();if(!$stock||$stock-total-$stock-frozen$quantity){returnfalse;}// 增加冻结Db::table(stocks)-where(product_id,$productId)-increment(frozen,$quantity);// 记录冻结流水幂等// ...returntrue;}finally{$lock-release();}}publicfunctionunfreeze(int$productId,int$quantity):bool{$lock$this-redis-lock(lock:inventory:.$productId,5);try{if(!$lock-get())returnfalse;$stockDb::table(stocks)-where(product_id,$productId)-first();if($stock-frozen$quantity){returnfalse;}Db::table(stocks)-where(product_id,$productId)-decrement(frozen,$quantity);returntrue;}finally{$lock-release();}}publicfunctiondeduct(int$productId,int$quantity):bool{$lock$this-redis-lock(lock:inventory:.$productId,5);try{if(!$lock-get())returnfalse;$stockDb::table(stocks)-where(product_id,$productId)-first();if($stock-total-$stock-frozen$quantity){returnfalse;}// 实际扣减减少 total 和 frozen因为之前已冻结Db::table(stocks)-where(product_id,$productId)-decrement(total,$quantity);Db::table(stocks)-where(product_id,$productId)-decrement(frozen,$quantity);returntrue;}finally{$lock-release();}}publicfunctiongetStock(int$productId):array{$stockDb::table(stocks)-where(product_id,$productId)-first();if(!$stock)return[available0,frozen0];return[available$stock-total-$stock-frozen,frozen$stock-frozen,];}}说明冻结流水的幂等记录未完整实现但已通过数据库唯一约束uk_saga_action预留实际 Saga 调用时会传入saga_id。依赖绑定App\JsonRpc\InventoryServiceInterface::classApp\JsonRpc\InventoryService::class,步骤 3配置 Consul 注册两个服务的config/autoload/services.php都需配置 Consul 驱动与用户服务相同。步骤 4在网关添加聚合接口为了前端能一次性获取商品列表含实时库存我们在hyperf-gateway中新增一个控制器ProductController通过 RPC 调用商品和库存服务进行数据聚合。首先在网关项目安装 RPC 客户端组件若未安装cd/var/www/hyperf-gatewaycomposerrequire hyperf/json-rpc配置消费者config/autoload/services.php添加 Product 和 Inventory 服务发现consumers[[nameProductService,registry[protocolconsul,addresshttp://consul:8500],],[nameInventoryService,registry[protocolconsul,addresshttp://consul:8500],],],在网关中创建app/Service/ProductAggregator.php注入两个 RPC 客户端?phpnamespaceApp\Service;useHyperf\RpcClient\Annotation\RpcClient;useApp\JsonRpc\ProductServiceInterface;useApp\JsonRpc\InventoryServiceInterface;classProductAggregator{#[RpcClient(name:ProductService)]privateProductServiceInterface$productService;#[RpcClient(name:InventoryService)]privateInventoryServiceInterface$inventoryService;publicfunctionlistProducts(int$page,int$pageSize):array{$result$this-productService-listProducts($page,$pageSize);$products$result[products];// 批量获取库存foreach($productsas$product){$stock$this-inventoryService-getStock($product[id]);$product[available_stock]$stock[available];}return[products$products,total$result[total]];}}创建app/Controller/ProductController.php?phpnamespaceApp\Controller;useApp\Service\ProductAggregator;useHyperf\HttpServer\Annotation\Controller;useHyperf\HttpServer\Annotation\RequestMapping;useHyperf\Di\Annotation\Inject;#[Controller(prefix:/products)]classProductControllerextendsAbstractController{#[Inject]privateProductAggregator$aggregator;#[RequestMapping(path:,methods:get)]publicfunctionindex(){$page(int)$this-request-input(page,1);$pageSize(int)$this-request-input(page_size,10);return$this-aggregator-listProducts($page,$pageSize);}}将此路由加入网关自身路由白名单避免被网关转发中间件拦截修改GatewayMiddleware中的$selfRoutes增加/products。四、启动与测试约 1 小时1. 启动所有服务# 分别进入 user, product, inventory 目录启动php bin/hyperf.php start确认 Consul UI 中出现ProductService和InventoryService且健康。2. 测试接口# 直接测试商品 RPCcurl-XPOST http://localhost:9503-HContent-Type: application/json\-d{jsonrpc:2.0,method:product/listProducts,params:[1,10],id:1}# 测试库存 RPCcurl-XPOST http://localhost:9504-HContent-Type: application/json\-d{jsonrpc:2.0,method:inventory/getStock,params:[1],id:2}# 通过网关聚合获取商品列表curlhttp://localhost:9500/products聚合接口应返回商品列表且每条商品带有available_stock。3. 测试库存冻结/解冻调用库存服务的冻结接口然后查询库存验证frozen增加available减少。高并发时可使用ab测试锁的效果。五、成果测试与记录测试清单检验项方法通过标准商品列表 RPC调用listProducts返回已插入的商品数据库存查询 RPC调用getStock返回当前可用和冻结库存网关聚合商品列表访问/products商品包含库存信息无延迟库存冻结调用freeze后查询库存frozen 增加available 减少库存解冻调用unfreeze后查询库存恢复分布式锁并发调用freeze无超卖库存正确Consul 服务发现Consul UI 查看两个新服务注册且健康商品创建RPC 创建商品自动同步库存需手动初始化库存或通过 Saga六、今日作业与学习产出提交代码将product-service和inventory-service项目及网关聚合代码提交。完善库存同步创建商品时商品服务通过 RPC 调用库存服务初始化库存记录。实现FreezeLog的幂等记录供 Saga 协调器使用。学习笔记画出商品列表请求的调用链客户端 → 网关 → 商品服务 → 库存服务RPC。总结库存服务使用分布式锁的场景和锁粒度选择。挑战任务为库存服务添加乐观锁版本控制对比 Redis 锁的性能。使用Kafka发布库存变更事件让商品服务缓存库存减少 RPC 调用。通过今天的学习电商核心系统的商品与库存中台已经搭建完毕它们通过 RPC 紧密协作为明天的订单服务打下了坚实基础。明天我们将实现订单服务并引入 Saga 事务串联起整个交易流程。