黄磊的博客 - 中间件
type: channel
link: https://www.huangleicole.com/category/middleware/
items: Elasticsearch实战：从零开始构建企业级日志检索系统
items: 消息队列选型：Kafka与RocketMQ在订单场景下的落地对比
items: Nacos配置中心原理浅析：长轮询是如何实现配置动态刷新的？
items: 利用RocketMQ实现分布式场景下的最终一致性：一个订单支付的例子

Elasticsearch实战：从零开始构建企业级日志检索系统
type: item
date: 2024-01-20T20:06:00+08:00
description: AI摘要 文章手把手演示用ELK搭建企业级日志系统：三节点ES集群保障高可用，Logstash通过grok/multiline把Java日志结构化，索引模板固化字段类型，最终在Kibana实现秒级可视化检索，将分散在数十台服务器的日志排查从“体力活”变为“侦探式”高效分析。 此内容根据文章生成，仅用于文章内容的解释与总结 作为开发者，我们最熟悉的调试方式就是看日志。当你的应用只有三五个实例时，SSH到服务器上用tail -f和grep还能勉强应付。但当微服务架构下，几十个应用分散在不同的机器上，一次请求的日志能散落在五六个地方时，传统的日志查看方式就变成了一场噩梦。一、架构选型：为什么是ELK/EFK？在开始之前，我们先明确核心架构。经典的ELK Stack包括：Elasticsearch：负责日志的存储、检索和聚合。它是我们系统的“大脑”。Logstash：负责日志的收集、解析、过滤和转发。它是数据的“加工厂”。Kibana：负责日志的可视化查询和展示。它是我们与系统交互的“窗口”。有时，人们会用更轻量的Fluentd或Filebeat替代Logstash，形成EFK Stack。在我们的实践中，由于需要对Java日志进行复杂的解析（比如解析堆栈异常），我们选择了功能更强大的Logstash。整个数据流非常清晰：应用日志文件 -> Logstash（收集、解析） -> Elasticsearch（存储） -> Kibana（查询展示）。二、第一步：部署与搭建的“坑”1. Elasticsearch集群配置：单机与集群的抉择一开始，我们在测试环境只部署了一个Elasticsearch节点。这很快带来了问题：只要这个节点一重启（比如调整配置后），整个日志系统就不可用了。解决方案：至少部署一个三节点集群。即使是在测试环境，我们也应该部署一个最小规模的高可用集群。以下是elasticsearch.yml的核心配置：# 节点1配置 cluster.name: company-logging-cluster # 集群名，所有节点必须一致 node.name: es-node-1 # 节点名，唯一 node.roles: [ master, data ] # 节点角色，兼具主节点和数据节点 network.host: 0.0.0.0 # 绑定地址 discovery.seed_hosts: ["es-node-1-ip", "es-node-2-ip", "es-node-3-ip"] # 种子节点列表 cluster.initial_master_nodes: ["es-node-1", "es-node-2", "es-node-3"] # 初始主节点关键点：discovery.seed_hosts：告诉节点在启动时去联系哪些节点来加入集群。cluster.initial_master_nodes：避免脑裂，明确指定哪些节点有资格在集群首次启动时参与主节点选举。角色分离：在生产环境，最好将master节点和data节点分离，但三节点模式下混搭是可行的。2. Logstash的管道（Pipeline）配置：理解数据流Logstash的配置文件是它的核心，采用“管道”概念，分为三个部分：input -> filter -> output。一个最基础的logstash.conf配置如下：input { # 从文件输入，这是最常见的日志来源 file { path => ["/app/logs/*.log"] # 监听哪些日志文件 start_position => "beginning" # 首次启动从文件开头读，还是"end"从结尾读 sincedb_path => "/dev/null" # 由于是测试，我们不记录读取位置。生产环境需指定一个文件。 } } filter { # 过滤器是空的，我们稍后来完善它 } output { # 输出到Elasticsearch elasticsearch { hosts => ["http://es-node-1:9200", "http://es-node-2:9200"] # ES集群地址 index => "application-logs-%{+YYYY.MM.dd}" # 定义索引名，按日期分割 } # 开发阶段非常有用：在控制台也输出一份，方便调试 stdout { codec => rubydebug } }启动Logstash后，你就能在Kibana上看到日志了，但所有日志都堆在message字段里，无法进行有效的搜索和筛选。这就是我们接下来要解决的核心问题。三、核心挑战：如何“解析”日志，而不仅仅是“存储”原始的日志就像一团乱麻。我们的目标是将它结构化，提取出时间戳、日志级别、…

消息队列选型：Kafka与RocketMQ在订单场景下的落地对比
type: item
date: 2023-12-01T19:36:00+08:00
description: AI摘要 订单系统需顺序、事务、延迟、高可靠。Kafka靠分区保序，事务、延迟需绕行，吞吐与流生态强；RocketMQ原生顺序队列、事务、18级延迟消息，开发运维简单。纯交易选RocketMQ，混合大数据场景选Kafka。 此内容根据文章生成，仅用于文章内容的解释与总结 在构建订单系统时，消息队列几乎成为了标配。从订单创建、库存扣减、积分累计到通知配送，每一个环节都可能通过消息队列进行异步化解耦。而当团队面临技术选型时，Apache Kafka和Apache RocketMQ往往是两个最常被考虑的选项。一、订单场景的核心诉求在开始对比前，我们必须先明确订单系统对消息队列的独特要求：顺序性：订单的状态流转必须严格有序。“已创建” -> “已支付” -> “已发货” -> “已完成”，这个顺序绝不能错乱。可靠性：每一笔订单消息都代表真实的交易，绝不能丢失。事务消息：创建订单和扣减库存需要保持最终一致性，这要求消息队列支持事务消息。延迟消息：自动取消超时未支付的订单，是订单系统的刚性需求。海量吞吐：大促期间，订单创建量是平时的数十甚至上百倍。下面，我们就围绕这几点，展开详细的对比。二、顺序消息：实现方式的根本差异场景：一个订单order_123，先后产生了CREATED、PAID、SHIPPED三条状态消息。消费端必须按此顺序处理。Kafka的实现方式与开发体验Kafka的顺序性基于分区（Partition） 保证。同一个Key的消息会被路由到同一个分区，而单个分区内的消息是有序的。代码实现：// 发送顺序消息：指定相同的key，确保进入同一分区 try (Producer<String, String> producer = new KafkaProducer<>(props)) { producer.send(new ProducerRecord<>("order_topic", orderId, "CREATED")); producer.send(new ProducerRecord<>("order_topic", orderId, "PAID")); // orderId相同，进入同一分区 producer.send(new ProducerRecord<>("order_topic", orderId, "SHIPPED")); }开发中遇到的“坑”：并发发送的乱序风险：如果使用异步发送（producer.send(record, callback)），即使Key相同，如果第一条消息发送失败重试，而第二条消息发送成功，就可能出现乱序。必须将max.in.flight.requests.per.connection设置为1，但这会严重影响吞吐量。分区数量变更的灾难：如果因为性能问题需要增加分区数，那么新的消息路由规则会改变，可能导致同一个订单的消息被路由到不同的新分区，顺序性被彻底破坏。在Kafka中，分区数一旦确定，几乎就是不可变的。消费端并行度的限制：为了保证顺序消费，一个分区只能被一个消费者线程处理。如果某个订单的处理非常耗时（比如涉及复杂的风控检查），它会阻塞该分区内所有其他订单的处理，形成“队头阻塞”。总结：Kafka的顺序性模型简单直接，但对运维和设计的提前规划要求极高，灵活性较差。RocketMQ的实现方式与开发体验RocketMQ专门提出了消息组（Message Group） 的概念，在Kafka分区的基础上做了更高层次的封装。代码实现：// RocketMQ发送顺序消息更加直观 Message msg1 = new Message("OrderTopic", "CREATED", orderId.getBytes()); Message msg2 = new Message("OrderTopic", "PAID", orderId.getBytes()); Message msg3 = new Message("OrderTopic", "SHIPPED", orderId.getBytes()); // 使用MessageQueueSelector，确保同一订单选择同一个消息队列 SendResult sendResult1 = producer.send(msg1, new SelectMessageQueueByHash(), orderId); SendResult sendResult2 = producer.send(msg2, new SelectMessageQueueByHash(), orderId); SendResult sendResult3 = producer.send(msg3, new SelectMessageQueueByHash(), orderId);消费端：…

Nacos配置中心原理浅析：长轮询是如何实现配置动态刷新的？
type: item
date: 2022-04-14T15:09:00+08:00
description: AI摘要 文章以一次配置未及时生效的雪崩事故切入，剖析Nacos长轮询原理：客户端30s长连接hold，服务端用异步队列+MD5比对即时回推；结合Spring @RefreshScope事件链完成动态刷新。给出调优、重试、灰度、监控、诊断等全套生产级方案，强调“长轮询≠简单轮询”，唯有深研原理与观测体系，才能避免“深夜告警”。 此内容根据文章生成，仅用于文章内容的解释与总结 我曾经以为配置中心就是简单的"发布-订阅"，直到那个深夜的线上告警——某个核心服务的超时配置没有及时生效，导致整个系统雪崩，我才真正理解长轮询背后的复杂性。一、从一次配置更新失败的事故说起1.1 事故现场还原我们先来看看当时的配置和代码：# 在Nacos中的服务配置 # Data ID: order-service.yaml # Group: DEFAULT_GROUP feign: client: config: default: connectTimeout: 5000 # 计划从2000改为5000 readTimeout: 10000 # 计划从5000改为10000服务中的配置读取代码：@RefreshScope @RestController @RequestMapping("/orders") @Slf4j public class OrderController { @Value("${feign.client.config.default.connectTimeout:2000}") private Integer connectTimeout; @Value("${feign.client.config.default.readTimeout:5000}") private Integer readTimeout; @Autowired private OrderFeignClient orderFeignClient; @GetMapping("/{orderId}") public Order getOrder(@PathVariable String orderId) { log.info("当前超时配置 - connectTimeout: {}, readTimeout: {}", connectTimeout, readTimeout); // 这里调用的Feign客户端应该使用上面的超时配置 return orderFeignClient.getOrder(orderId); } @PostConstruct public void init() { log.info("OrderController初始化完成，超时配置 - connectTimeout: {}, readTimeout: {}", connectTimeout, readTimeout); } }在Nacos控制台修改配置后，监控显示：23:00:00 - 配置修改 23:00:30 - 服务A配置生效 23:01:15 - 服务B配置生效 23:02:45 - 服务C配置仍未生效 -> 开始出现超时错误 23:05:00 - 手动重启服务C后配置生效问题很明显：​不同服务实例的配置刷新时间差异很大，有的实例根本没有及时刷新​。二、Nacos配置中心架构总览2.1 Nacos配置中心的整体架构在深入长轮询之前，我们先了解Nacos配置中心的整体架构：2.2 核心组件解析// Nacos客户端配置管理的核心类 public class ClientWorker implements Closeable { // 配置缓存管理器 private ConfigCacheService configCacheService; // 长轮询管理器 private LongPollingManager longPollingManager; // 配置监听器 private Map<String, List<ManagerListenerWrap>> listenerMap; // 执行长轮询任务 public void checkConfigInfo() { // 这是长轮询的入口 } } // 服务端处理长轮询的核心类 @RestController @RequestMapping("/v1/cs") public class ConfigController { @PostMapping("/listener") public String listener(HttpServletRequest request, HttpServletResponse response) { // 处理客户端的监听请求 return longPollingService.doP…

利用RocketMQ实现分布式场景下的最终一致性：一个订单支付的例子
type: item
date: 2022-02-23T15:36:00+08:00
description: AI摘要 文章以订单支付为例，演示如何用 RocketMQ 事务消息替代同步调用，解决“用户已扣款但订单状态未更新”的数据不一致问题：先发送 Half 消息并记录支付流水，再经本地事务提交或回查决定消息是否生效；消费端幂等更新订单，配合重试、死信、补偿、降级、监控等手段，实现高可靠最终一致性。 此内容根据文章生成，仅用于文章内容的解释与总结 我以为引入RocketMQ就能轻松解决分布式事务问题，直到那次深夜告警——用户付了钱但订单状态未更新，我才意识到分布式场景下的数据一致性远比想象中复杂。一、问题现场：从数据不一致到业务损失1.1 最初的同步调用架构先看看我们最初的问题架构：// 支付服务 - 最初的同步调用实现 @Service @Slf4j public class PaymentService { @Autowired private PaymentMapper paymentMapper; @Autowired private RestTemplate restTemplate; /** * 支付成功回调处理 * 问题：同步调用订单服务，网络异常时数据不一致 */ @Transactional public void handlePaymentCallback(PaymentCallbackDTO callback) { try { // 1. 记录支付流水 PaymentRecord payment = createPaymentRecord(callback); paymentMapper.insert(payment); // 2. 更新支付订单状态 updatePaymentOrderStatus(callback.getOrderId(), "PAID"); // 3. 同步调用订单服务更新订单状态 String orderServiceUrl = "http://order-service/orders/" + callback.getOrderId() + "/pay-success"; ResponseEntity<String> response = restTemplate.postForEntity( orderServiceUrl, callback, String.class ); if (!response.getStatusCode().is2xxSuccessful()) { throw new RuntimeException("订单状态更新失败"); } log.info("支付成功处理完成，订单号：{}", callback.getOrderId()); } catch (Exception e) { log.error("支付回调处理失败", e); // 这里抛出异常，整个事务回滚 throw new RuntimeException("支付处理失败", e); } } } // 订单服务 @RestController @RequestMapping("/orders") public class OrderController { @PostMapping("/{orderId}/pay-success") public ResponseEntity<?> handlePaySuccess(@PathVariable String orderId, @RequestBody PaymentCallbackDTO callback) { // 这里可能因为各种原因失败 boolean success = orderService.updateOrderStatus(orderId, "PAID", callback.getAmount()); if (!success) { return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build(); } // 触发后续业务：扣减库存、发放优惠券等 asyncProcessAfterPayment(orderId); return ResponseEntity.ok().build(); } }1.2 问题的根源分析​核心问题​：第三方支付已经扣款成功，但由于订单服务调用失败，导致支付服务事务回滚，支付记录丢失。用户付了钱，但系统里没有任何记录。二、RocketMQ事务消息：分布式事务的优雅解决方案2.1 事务消息的基本原理RocketMQ的事务消息机制分为两个阶段：2.2 事务消息的关键类解析// RocketMQ事务消息的核心接口 public interface TransactionListener { /** * 执行本地事务 * @param ms…