当前位置： 首页 > news >正文

消息队列设计：构建异步通信与系统解耦的实践指南

news 2026/5/31 23:49:18

消息队列设计：构建异步通信与系统解耦的实践指南

消息队列是分布式系统中实现异步通信和系统解耦的核心基础设施。它允许应用程序通过消息传递进行通信，而不需要等待响应，从而提高系统的可扩展性、可靠性和性能。在现代微服务架构中，消息队列扮演着至关重要的角色。本文将从消息队列的核心概念、设计模式、应用场景、可靠性保障等多个维度，全面介绍消息队列的设计与实践。

一、消息队列核心概念

消息队列是一种进程间通信机制，它允许发送者将消息放入队列，接收者从队列中取出消息进行处理。这种异步的通信方式能够有效解耦生产者和消费者，让系统各部分可以独立扩展和演进。

消息队列的核心概念包括：消息（Message）是传输的基本单位，包含消息头和消息体；生产者（Producer）负责发送消息到队列；消费者（Consumer）从队列接收并处理消息；队列（Queue）存储消息的FIFO数据结构；交换机（Exchange）根据路由规则分发消息到队列；绑定（Binding）定义交换机和队列之间的路由关系。理解这些核心概念是设计消息系统的基础。

┌─────────────────────────────────────────────────────────┐ │ 消息队列核心架构 │ ├─────────────────────────────────────────────────────────┤ │ │ │ Producer ──► Exchange ──► Binding ──► Queue ──► Consumer│ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Direct │ │ Topic │ │ Fanout │ │ │ │ 交换机 │ │ 交换机 │ │ 交换机 │ │ │ └─────────┘ └─────────┘ └─────────┘ │ │ │ │ Routing Key: order.created, payment.*, notification.#│ │ │ └─────────────────────────────────────────────────────────┘

二、消息队列选型与对比

当前主流的消息队列产品有RabbitMQ、Kafka、RocketMQ、ActiveMQ等。每种产品都有其特点和适用场景，选择合适的消息队列需要综合考虑业务需求、技术团队能力、运维成本等因素。

RabbitMQ适合中小型系统，特点是功能丰富、支持多种协议、社区活跃；Kafka适合大数据场景，特点是高吞吐量、持久化、分布式；RocketMQ是阿里巴巴开源的消息队列，适合电商等高并发场景；ActiveMQ是老牌消息队列，但性能相对较弱。

/** * 消息队列选型指南 */ public class MessageQueueSelectionGuide { public static void main(String[] args) { System.out.println("=== 消息队列选型对比 ===\n"); System.out.println("【RabbitMQ】"); System.out.println(" 特点:"); System.out.println(" - 功能丰富，支持多种消息模式"); System.out.println(" - 支持AMQP、STOMP、MQTT等多种协议"); System.out.println(" - 社区活跃，文档完善"); System.out.println(" - 管理界面友好"); System.out.println(" 适用场景:"); System.out.println(" - 业务消息队列"); System.out.println(" - 任务队列、异步处理"); System.out.println(" - 需要复杂路由的场景"); System.out.println(" 不适合:"); System.out.println(" - 超高吞吐量日志处理"); System.out.println(" - 海量消息堆积"); System.out.println(); System.out.println("【Kafka】"); System.out.println(" 特点:"); System.out.println(" - 极高吞吐量，单机可达百万级QPS"); System.out.println(" - 消息持久化，支持消息回溯"); System.out.println(" - 分布式架构，支持分区和副本"); System.out.println(" - 生态丰富（Kafka Streams、Connect等）"); System.out.println(" 适用场景:"); System.out.println(" - 日志收集与分析"); System.out.println(" - 实时流处理"); System.out.println(" - 大数据场景的消息管道"); System.out.println(" 不适合:"); System.out.println(" - 复杂的消息路由"); System.out.println(" - 金融级事务消息"); System.out.println(); System.out.println("【RocketMQ】"); System.out.println(" 特点:"); System.out.println(" - 高吞吐、高可靠"); System.out.println(" - 支持事务消息"); System.out.println(" - 延迟消息支持好"); System.out.println(" - 阿里生产验证"); System.out.println(" 适用场景:"); System.out.println(" - 电商交易系统"); System.out.println(" - 订单处理、库存扣减"); System.out.println(" - 需要事务保证的场景"); } }

三、消息队列设计模式

消息队列支持多种设计模式，每种模式都有其适用场景。理解这些模式有助于设计更好的消息系统。

点对点模式（Point-to-Point）是最基本的模式，消息被一个消费者消费后即被删除；发布订阅模式（Publish-Subscribe）允许多个消费者同时接收消息；工作队列模式（Work Queue）将任务分配给多个worker处理；请求响应模式（Request-Reply）实现RPC调用的异步化；Saga模式处理分布式事务；死信模式处理处理失败的消息。

import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.annotation.*; import org.springframework.stereotype.Service; import java.util.UUID; import java.util.concurrent.*; @Service public class MessagePatternExamples { @Autowired private RabbitTemplate rabbitTemplate; // ============ 点对点模式 ============ /** * 点对点模式：订单处理 * 一条消息只会被一个消费者处理 */ public void sendOrderMessage(Order order) { String queueName = "order.processing.queue"; rabbitTemplate.convertAndSend(queueName, order); System.out.println("订单消息已发送: " + order.getId()); } @RabbitListener(queues = "order.processing.queue") public void handleOrder(Order order) { System.out.println("处理订单: " + order.getId()); // 处理订单逻辑 } // ============ 发布订阅模式 ============ /** * 发布订阅模式：商品更新通知 * 多个消费者可以订阅同一类型的消息 */ public void publishProductUpdate(Product product) { String exchangeName = "product.update.exchange"; String routingKey = "product.updated." + product.getCategory(); rabbitTemplate.convertAndSend(exchangeName, routingKey, product); System.out.println("商品更新消息已发布: " + product.getId()); } // 搜索服务订阅 @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "search.product.update.queue"), exchange = @Exchange(name = "product.update.exchange"), key = "product.updated.*")) public void handleSearchIndex(Product product) { System.out.println("搜索服务：更新商品索引: " + product.getId()); } // 缓存服务订阅 @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "cache.product.update.queue"), exchange = @Exchange(name = "product.update.exchange"), key = "product.updated.*")) public void handleCacheUpdate(Product product) { System.out.println("缓存服务：更新商品缓存: " + product.getId()); } // 通知服务订阅 @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "notification.product.update.queue"), exchange = @Exchange(name = "product.update.exchange"), key = "product.updated.#")) public void handleNotification(Product product) { System.out.println("通知服务：发送商品更新通知: " + product.getId()); } // ============ 工作队列模式 ============ /** * 工作队列模式：图片处理 * 多个worker竞争处理消息，实现负载均衡 */ public void sendImageProcessTask(ImageTask task) { String queueName = "image.processing.queue"; rabbitTemplate.convertAndSend(queueName, task); } @RabbitListener(queues = "image.processing.queue", concurrency = "3-10") public void processImage(ImageTask task) { System.out.println("处理图片: " + task.getImageId()); // 图片处理逻辑 } // ============ 请求响应模式 ============ /** * 请求响应模式：异步RPC调用 */ private final Map<String, CompletableFuture<String>> pendingRequests = new ConcurrentHashMap<>(); public String asyncCall(String serviceName, String request) { String correlationId = UUID.randomUUID().toString(); String replyQueue = "rpc.reply." + correlationId; // 创建Future CompletableFuture<String> future = new CompletableFuture<>(); pendingRequests.put(correlationId, future); // 发送请求 rabbitTemplate.convertAndSend("rpc.exchange", serviceName, request, message -> { message.getMessageProperties().setCorrelationId(correlationId); message.getMessageProperties().setReplyTo(replyQueue); return message; }); // 设置超时 CompletableFuture<String> timeoutFuture = future.orTimeout(5, TimeUnit.SECONDS); timeoutFuture.exceptionally(ex -> { pendingRequests.remove(correlationId); return "Request timeout"; }); return future.join(); } @RabbitListener(queues = "rpc.reply.*") public void handleRpcReply(String response, org.springframework.amqp.core.Message message) { String correlationId = message.getMessageProperties().getCorrelationId(); CompletableFuture<String> future = pendingRequests.remove(correlationId); if (future != null) { future.complete(response); } } }

四、消息可靠性设计

消息可靠性是消息队列系统的生命线。在分布式环境中，网络故障、服务宕机等情况时有发生，需要通过各种机制保证消息不丢失。消息可靠性涉及三个环节：生产者到Broker的可靠性、Broker本身的可靠性、Broker到消费者的可靠性。

生产者可靠性通过事务消息、发布确认等机制保证；Broker可靠性通过消息持久化、镜像队列等机制保证；消费者可靠性通过手动确认、消费幂等性等机制保证。只有三个环节都做好，才能实现端到端的消息可靠性。

import com.rabbitmq.client.*; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.annotation.PostConstruct; import java.io.IOException; import java.util.UUID; import java.util.concurrent.TimeUnit; @Configuration public class ReliableMessageConfig { @Bean public RabbitTemplate reliableRabbitTemplate( CachingConnectionFactory connectionFactory) { RabbitTemplate template = new RabbitTemplate(connectionFactory); // 设置消息转换器 template.setMessageConverter(new Jackson2JsonMessageConverter()); // 启用发布确认 connectionFactory.setPublisherConfirmType( CachingConnectionFactory.ConfirmType.CORRELATED); connectionFactory.setPublisherReturns(true); // 设置Mandatory，确保消息能路由到队列 template.setMandatory(true); // 设置确认回调 template.setConfirmCallback((correlationData, ack, cause) -> { if (ack) { System.out.println("消息已确认: " + (correlationData != null ? correlationData.getId() : "unknown")); } else { System.out.println("消息未确认，原因: " + cause); // 这里应该重发消息或记录日志 handleMessageNotConfirmed(correlationData, cause); } }); // 设置返回回调 template.setReturnsCallback(returned -> { System.out.println("消息返回: 路由键=" + returned.getRoutingKey() + ", 交换机=" + returned.getExchange() + ", 响应码=" + returned.getReplyCode()); // 消息无法路由，需要处理 handleUnroutableMessage(returned); }); return template; } private void handleMessageNotConfirmed( RabbitTemplate.ConfirmCallback.CorrelationData correlationData, String cause) { // 记录日志或重发消息 if (correlationData != null) { System.err.println("消息 " + correlationData.getId() + " 未确认: " + cause); } } private void handleUnroutableMessage(RabbitTemplate.Returned returned) { // 处理无法路由的消息 System.err.println("消息无法路由到队列: " + "交换机=" + returned.getExchange() + ", 路由键=" + returned.getRoutingKey()); } } @Service public class ReliableProducer { @Autowired private RabbitTemplate rabbitTemplate; /** * 可靠消息发送 */ public void sendReliableMessage(Order order) { String correlationId = UUID.randomUUID().toString(); rabbitTemplate.convertAndSend("order.exchange", "order.created", order, message -> { message.getMessageProperties().setCorrelationId(correlationId); message.getMessageProperties().setDeliveryMode( MessageDeliveryMode.PERSISTENT); return message; }); System.out.println("可靠消息已发送, correlationId: " + correlationId); } } @Service public class ReliableConsumer { @Autowired private OrderService orderService; /** * 可靠消息消费 */ @RabbitListener(queues = "order.processing.queue") public void handleOrderMessage(Order order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) { try { // 业务处理 orderService.processOrder(order); // 手动确认 channel.basicAck(deliveryTag, false); } catch (Exception e) { System.err.println("处理订单失败: " + e.getMessage()); try { // 记录失败日志 logOrderProcessingFailure(order, e); // 拒绝消息，是否重新入队取决于错误类型 if (isRetryableError(e)) { // 可重试错误，重新入队 channel.basicNack(deliveryTag, false, true); } else { // 不可重试错误，发送到死信队列 channel.basicNack(deliveryTag, false, false); } } catch (IOException ioException) { ioException.printStackTrace(); } } } private boolean isRetryableError(Exception e) { // 判断是否为可重试错误 return e instanceof NetworkException || e instanceof TimeoutException; } private void logOrderProcessingFailure(Order order, Exception e) { // 记录失败日志，用于后续分析和处理 System.err.println("订单处理失败: orderId=" + order.getId() + ", error=" + e.getMessage()); } }

五、消息顺序与幂等性

消息顺序和幂等性是消息队列使用中的两个重要问题。消息顺序问题是指消息的处理顺序与发送顺序不一致，可能导致业务逻辑错误；幂等性问题是指同一条消息被重复消费时产生重复结果。

保证消息顺序的常用方法包括：使用单分区或按某个维度（如用户ID）分区；使用消息序列号；在消费端进行排序处理。保证幂等性的常用方法包括：使用消息ID去重；业务状态机判断；数据库唯一约束；分布式锁等。

@Service public class OrderMessageProcessing { @Autowired private OrderRepository orderRepository; private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet(); /** * 顺序消息处理 */ public void processOrderInSequence(OrderMessage message) { String orderId = message.getOrderId(); // 方案1：使用数据库锁保证顺序 synchronized (orderId.intern()) { Order order = orderRepository.findById(orderId) .orElseThrow(() -> new OrderNotFoundException(orderId)); switch (message.getEventType()) { case "ORDER_CREATED": processOrderCreated(order, message); break; case "ORDER_PAID": processOrderPaid(order, message); break; case "ORDER_SHIPPED": processOrderShipped(order, message); break; case "ORDER_COMPLETED": processOrderCompleted(order, message); break; } orderRepository.save(order); } } /** * 幂等性处理 */ public void processOrderIdempotent(OrderMessage message) { String messageId = message.getMessageId(); // 方案1：使用Set记录已处理的消息ID if (processedMessageIds.contains(messageId)) { System.out.println("消息已处理过，跳过: " + messageId); return; } try { // 业务处理 doProcessOrder(message); // 记录已处理的消息ID processedMessageIds.add(messageId); // 定期清理，防止内存溢出 if (processedMessageIds.size() > 100000) { cleanOldMessageIds(); } } catch (Exception e) { // 处理失败，不记录messageId，下次会重新处理 throw e; } } /** * 方案2：使用数据库唯一约束保证幂等 */ public void processOrderWithDatabaseIdempotency(OrderMessage message) { try { // 尝试插入消息记录，如果已存在则跳过 MessageRecord record = new MessageRecord(); record.setMessageId(message.getMessageId()); record.setProcessedAt(new Date()); record.setStatus("PROCESSED"); messageRecordRepository.save(record); // 插入成功，执行业务逻辑 doProcessOrder(message); } catch (DataIntegrityViolationException e) { // 唯一键冲突，说明消息已处理过 System.out.println("消息已处理过: " + message.getMessageId()); } } /** * 方案3：使用业务状态机保证幂等 */ public void processOrderWithStateMachine(OrderMessage message) { String orderId = message.getOrderId(); String eventType = message.getEventType(); Order order = orderRepository.findById(orderId) .orElseThrow(() -> new OrderNotFoundException(orderId)); switch (eventType) { case "ORDER_PAID": // 只有在PENDING状态下才处理 if (order.getStatus() == OrderStatus.PENDING) { order.setStatus(OrderStatus.PAID); orderRepository.save(order); } break; case "ORDER_SHIPPED": // 只有在PAID状态下才处理 if (order.getStatus() == OrderStatus.PAID) { order.setStatus(OrderStatus.SHIPPED); orderRepository.save(order); } break; case "ORDER_COMPLETED": // 只有在SHIPPED状态下才处理 if (order.getStatus() == OrderStatus.SHIPPED) { order.setStatus(OrderStatus.COMPLETED); orderRepository.save(order); } break; } } private void cleanOldMessageIds() { // 清理超过24小时的messageId processedMessageIds.clear(); } private void doProcessOrder(OrderMessage message) { // 实际的处理逻辑 } private void processOrderCreated(Order order, OrderMessage message) { // 处理订单创建 } private void processOrderPaid(Order order, OrderMessage message) { // 处理订单支付 } private void processOrderShipped(Order order, OrderMessage message) { // 处理订单发货 } private void processOrderCompleted(Order order, OrderMessage message) { // 处理订单完成 } }

六、消息队列最佳实践

在实际项目中使用消息队列，需要遵循一些最佳实践，以确保系统的稳定性、可维护性和性能。以下是经过实践验证的经验总结。

命名规范方面，队列、交换机、路由键应采用有意义的命名，如业务.功能.queue格式。配置管理方面，关键参数应可配置化，如重试次数、超时时间等。监控告警方面，应监控队列深度、消费延迟、死信队列等关键指标。容量规划方面，应根据业务量合理规划队列数量和资源使用。

import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQBestPracticesConfig { /** * 最佳实践1：合理的连接和通道配置 */ @Bean public CachingConnectionFactory connectionFactory() { CachingConnectionFactory factory = new CachingConnectionFactory(); // 生产服务器配置 factory.setHost("rabbitmq-host"); factory.setPort(5672); factory.setUsername("producer"); factory.setPassword("password"); factory.setVirtualHost("/"); // 连接池配置 factory.setChannelCacheSize(50); // 通道缓存大小 factory.setConnectionCacheSize(10); // 连接缓存大小 // 心跳配置 factory.setRequestedHeartbeat(60); // 自动恢复 factory.setAutomaticRecoveryEnabled(true); factory.setNetworkRecoveryInterval(10000); // 发布确认 factory.setPublisherConfirmType( CachingConnectionFactory.ConfirmType.CORRELATED); factory.setPublisherReturns(true); return factory; } /** * 最佳实践2：消费者配置优化 */ @Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( CachingConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(new Jackson2JsonMessageConverter()); // 预取数量 - 根据处理能力调整 factory.setPrefetchCount(100); // 并发消费者数 factory.setConcurrentConsumers(3); factory.setMaxConcurrentConsumers(10); // 手动确认模式 factory.setAcknowledgeMode( org.springframework.amqp.core.AcknowledgeMode.MANUAL); // 批量监听 factory.setBatchListener(true); factory.setBatchSize(10); factory.setReceiveTimeout(100L); // 默认消费者失败处理 factory.setDefaultRequeueRejected(false); return factory; } /** * 最佳实践3：RabbitTemplate配置 */ @Bean public RabbitTemplate rabbitTemplate( CachingConnectionFactory connectionFactory) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setMessageConverter(new Jackson2JsonMessageConverter()); // 使用JSON格式 template.setExchange("default.exchange"); template.setRoutingKey("default.key"); return template; } } /** * 最佳实践4：命名规范示例 */ public class NamingConventions { // 交换机命名：业务.功能.exchange public static final String ORDER_EXCHANGE = "order.exchange"; public static final String PAYMENT_EXCHANGE = "payment.exchange"; public static final String NOTIFICATION_EXCHANGE = "notification.exchange"; // 队列命名：业务.功能.queue public static final String ORDER_PROCESSING_QUEUE = "order.processing.queue"; public static final String ORDER_DEAD_LETTER_QUEUE = "order.dlq.queue"; public static final String PAYMENT_PROCESSING_QUEUE = "payment.processing.queue"; // 路由键命名：业务.事件.维度 public static final String ORDER_CREATED_KEY = "order.created"; public static final String ORDER_PAID_KEY = "order.paid"; public static final String ORDER_CANCELLED_KEY = "order.cancelled"; } /** * 最佳实践5：监控指标收集 */ @Service public class MessageQueueMonitoring { @Autowired private RabbitTemplate rabbitTemplate; private final MeterRegistry meterRegistry; public MessageQueueMonitoring(MeterRegistry meterRegistry) { this.meterRegistry = meterRegistry; } /** * 记录发送消息数 */ public void recordMessageSent(String queueName, int count) { meterRegistry.counter("rabbitmq.messages.sent", "queue", queueName).increment(count); } /** * 记录消费消息数 */ public void recordMessageConsumed(String queueName, int count) { meterRegistry.counter("rabbitmq.messages.consumed", "queue", queueName).increment(count); } /** * 记录消费延迟 */ public void recordConsumeLatency(String queueName, long latencyMs) { meterRegistry.timer("rabbitmq.consume.latency", "queue", queueName).record(latencyMs, TimeUnit.MILLISECONDS); } /** * 记录处理失败 */ public void recordProcessingFailure(String queueName, String errorType) { meterRegistry.counter("rabbitmq.processing.failure", "queue", queueName, "error", errorType).increment(); } }

总结

消息队列是构建现代分布式系统不可或缺的基础设施。通过本文的介绍，我们了解了消息队列的核心概念、设计模式、可靠性保障、顺序与幂等性处理以及最佳实践。

在实际应用中，选择合适的消息队列产品、设计合理的消息模式、保证消息的可靠性、处理顺序和幂等性问题，遵循最佳实践，是构建高质量消息系统的关键。同时，监控和容量规划也是保障系统稳定运行的重要环节。

http://www.jsqmd.com/news/925787/

相关文章：

  • 深圳南山专业搬家公司推荐 粤海电子设备搬运攻略 - 从来都是英雄出少年
  • Gemini多语言发布会策划全链路复盘（含欧盟GDPR话术库+亚太KOL分级激活清单）
  • 2026廊坊GEO服务商实力榜单推荐TOP5 专业选型与避坑全指南 - 余小铁
  • 我现在的这套系统和小龙虾有什么区别
  • Gemini文案生成不是“抄作业”：揭秘头部品牌如何用它实现个性化触达+实时动态优化
  • 如何永久备份微信聊天记录？WeChatMsg开源工具完整解决方案
  • 4. 机器翻译任务
  • 健康 检查
  • 深圳搬家公司家具拆装：熟练高效 全程无损 专业团队上门服务 - 从来都是英雄出少年
  • 大大降低token费用的方法----------先ocr然后给AI
  • 黄仁勋怒怼“AI 裁员甩锅”：真正危险的，不是 AI 抢饭碗，而是别人已经用 AI 拉开差距
  • 紧急！Gemini监测延迟超117秒？这6个服务器级配置正在 silently 拖垮你的响应时效
  • 手把手教你用老毛桃PE给全盘格式化的电脑重建引导分区（附详细图文）
  • 别再手动改乱码了！用convmv命令5分钟批量搞定Linux中文文件名编码转换
  • 数据库设计优化：从原理到实践的完整指南
  • 构建之法阅读笔记 09
  • Flutter 表单处理与验证详解：构建健壮的表单系统
  • 微服务拆分策略：从单体到分布式的演进之路
  • AgentScope2
  • 7个实战技巧让Playnite游戏库管理效率翻倍
  • 联想电脑F11一键还原丢了别慌！手把手教你用官方工具找回原厂系统（含Office激活）
  • 从‘/’目录开始：一次搞懂Linux根文件系统里那些‘神秘’的文件夹都是干嘛用的
  • 警惕“虚假增长陷阱”：Gemini用户质量衰减曲线首次披露，3类高危行为正在侵蚀LTV
  • Gemini企业级审计实战指南（含NIST SP 800-53映射表）
  • 保姆级教程：用戴尔生命周期控制器+U盘，给PowerEdge T640配置RAID并安装系统
  • P11363 [NOIP2024] 树的遍历
  • 改图片尺寸工具入门指南，新手使用调整大小实用攻略 - 软件工具教程方法
  • 架构演进之路：从单体到云原生的技术变革
  • 【Gemini系统维护权威指南】：20年SRE亲授3大避坑法则与5分钟应急响应流程
  • 从一次GCC编译崩溃，我搞懂了Linux的ulimit和文件描述符到底怎么管