服务间通信开发实践

微服务架构中，服务间通信是核心问题。本文从实战角度出发，详解同步调用与异步通信的区别、选型原则。

服务间通信分类

服务间通信 (Inter-Service Communication)
│
├── 🔹 同步通信（请求-响应模式）
│   ├── HTTP/REST（OpenFeign, WebClient, Spring MVC）
│   └── RPC（Dubbo, gRPC, Thrift, Hessian）
│
└── 🔹 异步通信（消息/事件驱动模式）
    ├── 消息队列（Kafka, RocketMQ, RabbitMQ）
    ├── 事件总线 / 流处理（EventBridge, Flink, Spring Cloud Stream）
    └── 响应式非阻塞 HTTP（WebFlux + WebClient）

核心概念对比

RPC vs HTTP/REST

RPC（Remote Procedure Call）

• 是一种通信范式（设计思想），不是协议，也不是框架
• 目的是让远程服务调用看起来像本地方法调用一样
• 通信模型：面向接口/方法调用，传输协议可插拔（TCP / HTTP/2 / 自定义）

HTTP/REST

• 通信模型：面向资源/URL，数据格式为文本
• 框架基于库实现

RPC 框架特点

传统 RPC 框架：默认走 TCP + 自定义二进制协议

• 框架设计故意绕过 HTTP，减少头部开销、降低序列化损耗、提升吞吐
• 适合内网高性能场景
• 代表：Dubbo（自定义协议）、Thrift

现代 RPC 框架（如 gRPC、Dubbo 3 Triple）：复用 HTTP/2 的传输层能力

• 利用多路复用、头部压缩、双向流等特性
• 上层语义、IDL 契约、序列化（Protobuf）完全独立于传统 HTTP/REST

具体框架分层

┌─────────────────────────────────────────────────────┐
│                      框架分层                         │
├─────────────────────────────────────────────────────┤
│                                                      │
│  RPC 框架                                            │
│  ├─ 现代 RPC: gRPC, Dubbo 3 Triple                  │
│  └─ 传统 RPC: Dubbo (自定义协议), Thrift            │
│                                                      │
│  HTTP 客户端（框架）                                  │
│  ├─ 声明式: OpenFeign, Retrofit                     │
│  ├─ 同步模板: RestTemplate (已 Deprecated)           │
│  └─ 响应式: WebClient (推荐)                         │
│                                                      │
│  HTTP 库（底层依赖）                                  │
│  └─ OkHttp, HttpClient (Apache)                     │
│                                                      │
└─────────────────────────────────────────────────────┘

关键说明：

• OkHttp / HttpClient：HTTP 客户端库，负责连接、请求发送、响应接收。其他上层工具如 RestTemplate/Feign/Retrofit 都依赖这些库实现
• OpenFeign：声明式 HTTP 客户端（接口 + 注解，运行时通过 JDK 动态代理生成实现类），传输层依赖 HttpClient
• RestTemplate：Spring 提供的同步 HTTP 客户端模板，底层依赖 HttpClient，官方已标记 @Deprecated，推荐使用 WebClient（响应式）或 RestClient

第一部分：同步通信

1.1 HTTP 客户端：从 RestTemplate 到 OpenFeign

1.1.1 HTTP 库 vs HTTP 框架

HTTP 库（底层）：
├─ OkHttp       # 高性能，支持连接池、HTTP/2
└─ HttpClient   # Apache HTTP Components

HTTP 框架（上层）：
├─ RestTemplate  # 同步模板，已Deprecated
├─ OpenFeign     # 声明式接口，依赖HttpClient
└─ WebClient     # 响应式非阻塞，推荐使用

1.1.2 RestTemplate（已Deprecated）

// 问题：硬编码 URL，无超时，无负载均衡
RestTemplate restTemplate = new RestTemplate();
String url = "http://localhost:8080/order-service/orders/" + orderId;
return restTemplate.getForObject(url, Order.class);

1.1.3 OpenFeign：声明式 HTTP 客户端

OpenFeign 将 HTTP 调用伪装成接口调用，像调用本地方法一样调用远程服务。

@FeignClient(name = "order-service", path = "/orders")
public interface OrderFeignClient {

    @GetMapping("/{orderId}")
    Order getOrderById(@PathVariable("orderId") Long orderId);

    @PostMapping
    Order createOrder(@RequestBody CreateOrderRequest request);
}

feign:
  client:
    config:
      default:
        connectTimeout: 5000      # 连接超时
        readTimeout: 10000        # 读取超时
      order-service:
        connectTimeout: 3000
        readTimeout: 5000
  circuitbreaker:
    enabled: true

核心配置：

重试机制： 默认在 IOException（连接超时、读取超时）、502/503/504 时自动重试。

请求拦截器（TraceId 传递）：

@Component
public class FeignRequestInterceptor implements RequestInterceptor {
    @Override
    public void apply(RequestTemplate request) {
        String traceId = MDC.get("traceId");
        if (traceId != null) {
            request.header("X-Trace-Id", traceId);
        }
    }
}

1.1.4 实战：Feign + Ribbon 负载均衡

Ribbon 是客户端负载均衡器，与 Feign 无缝集成。

order-service:
  ribbon:
    listOfServers: localhost:8080,localhost:8081,localhost:8082
    ConnectTimeout: 3000
    ReadTimeout: 5000
    MaxAutoRetries: 2          # 当前实例重试次数
    MaxAutoRetriesNextServer: 1  # 切换实例重试次数

重试公式： 总重试次数 = (MaxAutoRetries + 1) × (MaxAutoRetriesNextServer + 1)

1.1.5 实战：Feign + Sentinel 熔断降级

@FeignClient(name = "order-service", path = "/orders",
             fallback = OrderFeignClientFallback.class)
public interface OrderFeignClient {
    @GetMapping("/{orderId}")
    Order getOrderById(@PathVariable("orderId") Long orderId);
}

// 降级处理
@Component
public class OrderFeignClientFallback implements OrderFeignClient {
    @Override
    public Order getOrderById(Long orderId) {
        Order order = new Order();
        order.setId(orderId);
        order.setStatus("FETCH_FAILED");
        return order;
    }
}

1.2 RPC 框架对比

1.2.1 传统 RPC：Dubbo

Dubbo 基于 TCP + 自定义二进制协议，低延迟、高吞吐，适合内部 Java 微服务。

public interface OrderService {
    @DubboReference(timeout = 3000, retries = 2)
    Order getOrderById(Long orderId);

    @DubboReference(loadbalance = "roundrobin")
    List<Order> getOrdersByUserId(Long userId);
}

dubbo:
  registry:
    address: nacos://localhost:8848
  protocol:
    name: dubbo
    port: 20880
    threads: 200

Dubbo 内核：

• 协议：自定义 Dubbo 协议（TCP），支持多协议扩展
• 序列化：Hessian / JSON / Kryo / Protobuf
• 负载均衡：Random / RoundRobin / LeastActive / 一致性 Hash
• 注册发现：ZooKeeper / Nacos / Consul

1.2.2 现代 RPC：gRPC

gRPC 基于 HTTP/2 + Protocol Buffers，支持跨语言，适合外部服务调用。

// order.proto
syntax = "proto3";
service OrderService {
  rpc GetOrder(GetOrderRequest) returns (Order);
}
message GetOrderRequest { int64 order_id = 1; }
message Order {
  int64 id = 1;
  int64 user_id = 2;
  string status = 3;
}

// gRPC 客户端
ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
OrderServiceBlockingStub stub = OrderServiceGrpc.newBlockingStub(channel);
Order order = stub.getOrder(GetOrderRequest.newBuilder().setOrderId(orderId).build());

gRPC 核心优势：

1.2.3 Dubbo vs gRPC vs Thrift

1.3 响应式非阻塞：WebFlux + WebClient

1.3.1 什么是响应式

同步阻塞：线程发起 I/O 后等待响应，线程阻塞 异步非阻塞：线程发起 I/O 后可继续处理其他任务，I/O 完成时回调通知

同步阻塞模型：
┌────────┐    请求     ┌────────┐
│ Client │ ────────→ │ Server │
│  Thread│ ←─────── │        │
│  阻塞  │    响应   │        │
└────────┘           └────────┘

异步非阻塞模型：
┌────────┐    请求     ┌────────┐
│ Client │ ────────→ │ Server │
└────────┘           └────────┘
     ↑    回调通知
     │
     └── 可处理其他任务

1.3.2 WebClient：响应式 HTTP 客户端

WebClient 是 Spring 5 提供的响应式 HTTP 客户端，推荐替代 RestTemplate。

// 声明式创建
WebClient webClient = WebClient.create("http://order-service");

// 带超时配置
WebClient webClient = WebClient.builder()
    .baseUrl("http://order-service")
    .clientConnector(new ReactorClientHttpConnector(
        HttpClient.create().responseTimeout(Duration.ofSeconds(10))
    ))
    .build();

// GET 请求
Mono<Order> orderMono = webClient.get()
    .uri("/orders/{orderId}", orderId)
    .header("X-Trace-Id", traceId)
    .retrieve()
    .bodyToMono(Order.class);

// POST 请求
Mono<Order> orderMono = webClient.post()
    .uri("/orders")
    .bodyValue(createOrderRequest)
    .retrieve()
    .bodyToMono(Order.class);

1.3.3 WebFlux：响应式 Web 框架

WebFlux 是 Spring 5 的响应式 Web 框架，基于 Netty（非 Servlet），支持完全异步非阻塞。

@RestController
public class OrderController {

    private final WebClient webClient;

    public OrderController(WebClient.Builder builder) {
        this.webClient = builder.baseUrl("http://order-service").build();
    }

    // 异步返回 Mono 或 Flux
    @GetMapping("/orders/{orderId}")
    public Mono<Order> getOrder(@PathVariable Long orderId) {
        return webClient.get()
            .uri("/orders/{orderId}", orderId)
            .retrieve()
            .bodyToMono(Order.class);
    }

    // 响应式调用链
    public Mono<Order> getOrderWithItems(Long orderId) {
        return webClient.get()
            .uri("/orders/{orderId}", orderId)
            .retrieve()
            .bodyToMono(Order.class)
            .flatMap(order -> webClient.get()
                .uri("/orders/{orderId}/items", orderId)
                .retrieve()
                .bodyToMono<List<OrderItem>>()
                .map(items -> { order.setItems(items); return order; })
            );
    }
}

1.3.4 响应式适用场景

第二部分：异步通信

2.1 MQ 三强对比：Kafka / RocketMQ / RabbitMQ

2.1.1 选型决策表

2.1.2 Kafka：高吞吐量日志流

核心概念： Topic / Partition / Consumer Group / Offset

// 生产者
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("topic", "key", "value"), (metadata, ex) -> {
    if (ex != null) { /* 处理失败 */ }
});

// 消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processMessage(record.value());
        consumer.commitSync();  // 手动提交
    }
}

2.1.3 RocketMQ：事务消息专家

RocketMQ 原生支持事务消息，是金融级可靠消息的首选。

// 事务生产者
TransactionMQProducer producer = new TransactionMQProducer("order-producer-group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 本地事务：创建订单
        orderService.createOrder(parseOrder(msg));
        return LocalTransactionState.COMMIT_MESSAGE;
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查：检查订单是否存在
        return orderService.isOrderCreated(parseOrderId(msg))
            ? LocalTransactionState.COMMIT_MESSAGE
            : LocalTransactionState.UNKNOW;
    }
});
producer.sendMessageInTransaction(message, null);

2.1.4 RabbitMQ：灵活路由

RabbitMQ 基于 AMQP，以灵活路由和可靠性著称。

核心概念： Exchange（路由）/ Queue（存储）/ Binding（绑定）

// 生产者：开启确认
channel.confirmSelect();
channel.basicPublish("exchange", "routing-key",
    MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
channel.waitForConfirms();  // 同步确认

// 消费者：手动 ACK
channel.basicConsume("queue", false, (tag, delivery) -> {
    try {
        processMessage(new String(delivery.getBody()));
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

2.2 可靠投递：生产端 → 消费端全链路

2.2.1 生产端可靠性

问题： 消息发送过程中丢失（网络故障、Broker 宕机）

方案 1：消息持久化 + 发布确认

rocketmq:
  producer:
    retryTimesWhenSendFailed: 3
    sendMessageTimeout: 10000
    acks: 2  # 所有 Follower 确认

方案 2：事务消息（见 2.3）

2.2.2 消费端可靠性

问题： 消费者拉取后处理失败，消息丢失

方案 1：手动 ACK

@RabbitListener(queues = "order-queue", ackMode = "MANUAL")
public void handleOrder(Message message, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
    try {
        processMessage(new String(message.getBody()));
        channel.basicAck(deliveryTag, false);  // 确认
    } catch (Exception e) {
        channel.basicNack(deliveryTag, false, true);  // 拒绝，重新入队
    }
}

方案 2：消费幂等

String key = message.getOrderId() + "-" + message.getMessageId();
if (deduplicateMapper.exists(key)) { return; }  // 已处理，跳过
processMessage(message.getOrder());
deduplicateMapper.save(key, LocalDateTime.now().plusDays(7));

方案 3：死信队列

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("order.dlx.queue")
        .withArgument("x-dead-letter-exchange", "order.dlx.exchange")
        .withArgument("x-dead-letter-routing-key", "order.dead")
        .build();
}

2.2.3 可靠投递完整链路

生产端                        MQ Broker                      消费端
  │                              │                              │
  ├─ 事务消息（本地+MQ回调）      │                              │
  ├─ 持久化刷盘                   │                              │
  ├─ 主从同步 ACK                │                              │
  │                              │←── 拉取消息 ────────────────  │
  │                              │                              │
  │                              │─── 手动 ACK ────────────────→│
  │                              │  （成功后才确认）            │
  ▼                              ▼                              ▼

2.3 事务消息：RocketMQ 实战

2.3.1 为什么需要事务消息

传统本地事务 + MQ 发送无法保证原子性：

// 问题：订单创建成功，但消息已发送，本地事务回滚了
@Transactional
public void createOrder(Order order) {
    orderMapper.insert(order);           // 本地事务成功
    mqProducer.send("order-topic", order);  // 消息已发送，无法撤回
    throw new RuntimeException("库存扣减失败");  // 本地事务回滚
}

事务消息：消息发送与本地事务绑定，要么同时成功，要么同时回滚。

2.3.2 事务消息原理

┌─────────────────────────────────────────────────────────────────┐
│  生产者           RocketMQ              消费者                    │
│     │                 │                   │                    │
│     ├─ 发送 Half Msg ─→│                   │                    │
│     │←─ 落盘成功 ──────┤                   │                    │
│     │                 │                   │                    │
│     ├─ 执行本地事务   │                   │                    │
│     │                 │                   │                    │
│     ├─ 提交/回滚 ───→│                   │                    │
│     │                 │                   │                    │
│     │        [COMMIT] ────────────────→   │  (对消费者可见)     │
│     │        [ROLLBACK] (删除消息)        │                    │
│     │                 │                   │                    │
│     │        [定时回查本地事务]           │                    │
│     │←────────────────┤                   │                    │
└─────────────────────────────────────────────────────────────────┘

关键机制：

1. Half Message：预处理消息，本地事务未提交前，对消费者不可见
2. 本地事务执行：业务操作与消息发送在同一事务中
3. Commit/Rollback：根据本地事务结果提交或回滚
4. 事务回查：超时未确认时，MQ 主动回查本地事务状态

2.3.3 实战：订单系统事务消息

@Service
@Slf4j
public class OrderTransactionService {

    @Autowired
    private OrderMapper orderMapper;

    public void createOrderWithTransaction(CreateOrderRequest request) {
        Order order = new Order();
        order.setId(SnowflakeIdGenerator.nextId());
        order.setUserId(request.getUserId());
        order.setAmount(request.getAmount());

        Message message = new Message("order-topic", "create-order",
            order.getId().toString(), JSON.toJSONString(order).getBytes());

        transactionProducer.sendMessageInTransaction(message, new TransactionExecuter() {
            @Override
            public LocalTransactionState executeLocalTransactionPhase(Message msg, Object arg) {
                try {
                    orderMapper.insert(order);
                    return LocalTransactionState.COMMIT_MESSAGE;
                } catch (Exception e) {
                    log.error("本地事务失败", e);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
            @Override
            public LocalTransactionState checkLocalTransactionPhase(MessageExt msg) {
                Long orderId = Long.parseLong(msg.getKeys());
                return orderMapper.selectById(orderId) != null
                    ? LocalTransactionState.COMMIT_MESSAGE
                    : LocalTransactionState.UNKNOW;
            }
        });
    }
}

2.4 最终一致性：Saga + 可靠消息

2.4.1 强一致性 vs 最终一致性

2.4.2 Saga 模式：Choreography vs Orchestration

Choreography（事件驱动编排）：

订单服务 ──── 创建订单成功 ────→ 库存服务 ──── 扣减成功 ────→ 支付服务
                                    ←──────────────────────────── 支付成功

// 事件驱动：服务间通过事件相互调用
@EventListener
public void handleOrderCreated(OrderCreatedEvent event) {
    eventPublisher.publish(new InventoryDeductEvent(orderId, items));
}

Orchestration（中心编排）：

// Saga Orchestrator：统一控制补偿
public void executeOrderSaga(CreateOrderRequest request) {
    String sagaId = UUID.randomUUID().toString();
    try {
        orderService.createOrder(request);
        inventoryService.deductStock(orderId, items);
        paymentService.pay(orderId, amount);
    } catch (Exception e) {
        // 逆序补偿
        paymentService.refund(orderId);
        inventoryService.restoreStock(orderId);
        orderService.cancelOrder(orderId);
    }
}

2.4.3 可靠消息 + 定时对账

服务 A                    MQ                      服务 B
  │                        │                        │
  ├─ 本地事务              │                        │
  ├─ 消息入库（待发送）     │                        │
  │                        │─── 定时扫描重试 ────→ │
  │                        │                        │
  │←───────────────────────│←─── 消费成功 ────────┤
  │                        │                        │
  │               [定时对账：检查双方状态]           │

2.5 事件总线 / 流处理

2.5.1 EventBridge（阿里云）

云原生事件总线，支持事件路由、过滤、转换。

// 发布事件
EventBridgeClient eventBridgeClient = new EventBridgeClient();
CloudEvent cloudEvent = CloudEventBuilder.builder()
    .withSource("order-service")
    .withType("order.created")
    .withData(JSON.toJSONString(order).getBytes())
    .build();
eventBridgeClient.putEvent(cloudEvent);

// 订阅规则
eventBridgeClient.createRule("order-topic-rule",
    FilterExpression.builder().equal("type", "order.created").build(),
    "order-service");

2.5.2 Spring Cloud Stream

统一 MQ 抽象层，支持 Kafka / RocketMQ / RabbitMQ 自动切换。

// 生产者
@Output("order-output")
interface OrderSource {
    void sendOrder(Order order);
}

// 消费者
@Input("order-input")
interface OrderSink {
    @StreamListener("order-input")
    void handleOrder(Order order);
}

第三部分：选型决策

3.1 同步 vs 异步判断

需要调用远程服务吗？
│
├─ 是 ──→ 需要同步等待返回？
│           │
│           ├─ 是 ──→ 内网 Java 微服务？
│           │           │
│           │           ├─ 是 ──→ Dubbo
│           │           │
│           │           └─ 否 ──→ HTTP/2 场景？
│           │                       │
│           │                       ├─ 是 ──→ gRPC
│           │                       │
│           │                       └─ 否 ──→ OpenFeign / WebClient
│           │
│           └─ 否 ──→ 异步可接受？
│                       │
│                       ├─ 事务消息需求？
│                       │       │
│                       │       ├─ 是 ──→ RocketMQ 事务消息
│                       │       │
│                       │       └─ 否 ──→ 高吞吐？
│                       │               │
│                       │               ├─ Kafka
│                       │               │
│                       │               └─ RabbitMQ
│                       │
│                       └─ 低延迟实时？
│                               │
│                               └─ WebFlux + WebClient
│
└─ 否 ──→ 事件驱动：MQ + 事件总线

3.2 关键配置规范

Feign 配置

feign:
  client:
    config:
      default:
        connectTimeout: 5000
        readTimeout: 10000
        retryer: Retryer.Default
  circuitbreaker:
    enabled: true

RocketMQ 生产端配置

rocketmq:
  producer:
    group: ${service}-producer
    sendMessageTimeout: 10000
    retryTimesWhenSendFailed: 3

Kafka 生产端配置

spring:
  kafka:
    producer:
      retries: 3
      acks: all          # 所有 ISR 确认
      batch-size: 16384
      linger.ms: 5

RabbitMQ 消费端配置

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 10
        retry:
          enabled: true
          max-attempts: 3

3.3 监控埋点

TraceId 传递（Feign）：

@Component
public class FeignTraceInterceptor implements RequestInterceptor {
    @Override
    public void apply(RequestTemplate request) {
        String traceId = MDC.get("traceId");
        if (traceId != null) {
            request.header("X-Trace-Id", traceId);
        }
    }
}

3.4 故障排查 Checklist

同步调用（Feign）

异步调用（MQ）

附录

A. 术语表

📚 关联文档

• 系统可观测性实践
• HTTP 与网络协议
• 分布式锁实践
• z-本地事务与分布式事务